# test_system_upgrade.py - unit tests for system-upgrade plugin import system_upgrade from system_upgrade import PLYMOUTH, CliError import os import tempfile import shutil import gettext from dnf.callback import (PKG_CLEANUP, PKG_DOWNGRADE, PKG_INSTALL, PKG_OBSOLETE, PKG_REINSTALL, PKG_REMOVE, PKG_UPGRADE, PKG_VERIFY, TRANS_POST) import unittest from tests.support import mock patch = mock.patch @patch('system_upgrade.call', return_value=0) class PlymouthTestCase(unittest.TestCase): def setUp(self): self.ply = system_upgrade.PlymouthOutput() self.msg = "Hello, plymouth." self.msg_args = (PLYMOUTH, "display-message", "--text", self.msg) def test_ping(self, call): self.ply.ping() call.assert_called_once_with((PLYMOUTH, "--ping")) self.assertTrue(self.ply.alive) def test_ping_when_dead(self, call): call.return_value = 1 self.ply.ping() self.assertFalse(self.ply.alive) call.return_value = 0 self.ply.ping() self.assertEqual(call.call_count, 2) self.assertTrue(self.ply.alive) def test_message(self, call): self.ply.message(self.msg) call.assert_called_once_with(self.msg_args) def test_hide_message(self, call): messages = ("first", "middle", "BONUS", "last") for m in messages: self.ply.message(m) def hidem(m): return mock.call((PLYMOUTH, "hide-message", "--text", m)) def dispm(m): return mock.call((PLYMOUTH, "display-message", "--text", m)) m1, m2, m3, m4 = messages call.assert_has_calls([ dispm(m1), hidem(m1), dispm(m2), hidem(m2), dispm(m3), hidem(m3), dispm(m4), ]) def test_message_dupe(self, call): self.ply.message(self.msg) self.ply.message(self.msg) call.assert_called_once_with(self.msg_args) def test_message_dead(self, call): call.return_value = 1 self.ply.message(self.msg) self.assertFalse(self.ply.alive) self.ply.message("not even gonna bother") call.assert_called_once_with(self.msg_args) def test_progress(self, call): self.ply.progress(27) call.assert_called_once_with( (PLYMOUTH, "system-update", "--progress", str(27))) @patch('system_upgrade.check_output', return_value="this plymouth does support --system-upgrade mode") def test_mode(self, check_output, call): self.ply.set_mode() call.assert_called_once_with((PLYMOUTH, "change-mode", "--system-upgrade")) @patch('system_upgrade.check_output', return_value="this plymouth doesn't support system-upgrade mode") def test_mode_no_system_upgrade_plymouth(self, check_output, call): self.ply.set_mode() call.assert_called_once_with((PLYMOUTH, "change-mode", "--updates")) def test_mode_no_plymouth(self, call): call.side_effect = OSError(2, 'No such file or directory') self.ply.set_mode() self.assertFalse(self.ply.alive) @patch('system_upgrade.call', return_value=0) class PlymouthTransactionProgressTestCase(unittest.TestCase): actions = (PKG_CLEANUP, PKG_DOWNGRADE, PKG_INSTALL, PKG_OBSOLETE, PKG_REINSTALL, PKG_REMOVE, PKG_UPGRADE, PKG_VERIFY, TRANS_POST) # pylint: disable=protected-access def setUp(self): system_upgrade.Plymouth = system_upgrade.PlymouthOutput() self.display = system_upgrade.PlymouthTransactionProgress() self.pkg = "testpackage" def test_display(self, call): for action in self.actions: self.display.progress(self.pkg, action, 0, 100, 1, 1000) msg = self.display._fmt_event(self.pkg, action, 1, 1000) # updating plymouth display means two plymouth calls call.assert_has_calls([ mock.call((PLYMOUTH, "system-update", "--progress", "0")), mock.call((PLYMOUTH, "display-message", "--text", msg)) ], any_order=True) def test_filter_calls(self, call): action = PKG_INSTALL # first display update -> set percentage and text self.display.progress(self.pkg, action, 0, 100, 1, 1000) msg1 = self.display._fmt_event(self.pkg, action, 1, 1000) call.assert_has_calls([ mock.call((PLYMOUTH, "system-update", "--progress", "0")), mock.call((PLYMOUTH, "display-message", "--text", msg1)), ]) # event progress on the same transaction item. # no new calls to plymouth because the percentage and text don't change for te_cur in range(1, 100): self.display.progress(self.pkg, action, te_cur, 100, 1, 1000) call.assert_has_calls([ mock.call((PLYMOUTH, "system-update", "--progress", "0")), mock.call((PLYMOUTH, "display-message", "--text", msg1)), ]) # new item: new message ("[2/1000] ..."), but percentage still 0.. self.display.progress(self.pkg, action, 0, 100, 2, 1000) # old message hidden, new message displayed. no new percentage. msg2 = self.display._fmt_event(self.pkg, action, 2, 1000) call.assert_has_calls([ mock.call((PLYMOUTH, "system-update", "--progress", "0")), mock.call((PLYMOUTH, "display-message", "--text", msg1)), mock.call((PLYMOUTH, "hide-message", "--text", msg1)), mock.call((PLYMOUTH, "display-message", "--text", msg2)), ]) TESTLANG = "zh_CN" TESTLANG_MO = "po/%s.mo" % TESTLANG @unittest.skipUnless(os.path.exists(TESTLANG_MO), "make %s first" % TESTLANG_MO) # @unittest.skip("There is no translation yet to system-upgrade") class I18NTestCaseBase(unittest.TestCase): @classmethod @unittest.skip("There is no translation yet to system-upgrade") def setUpClass(cls): cls.localedir = tempfile.mkdtemp(prefix='system_upgrade_test_i18n-') cls.msgdir = os.path.join(cls.localedir, TESTLANG + "/LC_MESSAGES") cls.msgfile = "dnf-plugins-extras" + ".mo" os.makedirs(cls.msgdir) shutil.copy2(TESTLANG_MO, os.path.join(cls.msgdir, cls.msgfile)) @classmethod def tearDownClass(cls): shutil.rmtree(cls.localedir) def setUp(self): self.t = gettext.translation("dnf-plugins-extras", self.localedir, languages=[TESTLANG], fallback=True) self.gettext = self.t.gettext class I18NTestCase(I18NTestCaseBase): @unittest.skip("There is no translation yet to system-upgrade") def test_selftest(self): self.assertIn(self.msgfile, os.listdir(self.msgdir)) self.assertIn(TESTLANG, os.listdir(self.localedir)) t = gettext.translation("dnf-plugins-extras", self.localedir, languages=[TESTLANG], fallback=False) info = t.info() self.assertIn("language", info) self.assertEqual(info["language"], TESTLANG.replace("_", "-")) @unittest.skip("There is no translation yet to system-upgrade") def test_fallback(self): msg = "THIS STRING DOES NOT EXIST" trans_msg = self.gettext(msg) self.assertEqual(msg, trans_msg) @unittest.skip("There is no translation yet to system-upgrade") def test_translation(self): msg = "the color of the sky" trans_msg = self.gettext(msg) self.assertNotEqual(msg, trans_msg) class StateTestCase(unittest.TestCase): @classmethod def setUpClass(cls): cls.statedir = tempfile.mkdtemp(prefix="system_upgrade_test_state-") cls.StateClass = system_upgrade.State def setUp(self): self.state = self.StateClass(os.path.join(self.statedir, "state")) def test_bool_value(self): with self.state: self.state.distro_sync = True del self.state self.state = self.StateClass(os.path.join(self.statedir, "state")) self.assertIs(self.state.distro_sync, True) @classmethod def tearDownClass(cls): shutil.rmtree(cls.statedir) class UtilTestCase(unittest.TestCase): def setUp(self): self.tmpdir = tempfile.mkdtemp(prefix='system_upgrade_test_util-') self.dirs = ["dir1", "dir2"] self.files = ["file1", "dir2/file2"] for d in self.dirs: os.makedirs(os.path.join(self.tmpdir, d)) for f in self.files: with open(os.path.join(self.tmpdir, f), 'wt') as fobj: fobj.write("hi there\n") def test_self_test(self): for d in self.dirs: self.assertTrue(os.path.isdir(os.path.join(self.tmpdir, d))) for f in self.files: self.assertTrue(os.path.exists(os.path.join(self.tmpdir, f))) def test_clear_dir(self): self.assertTrue(os.path.isdir(self.tmpdir)) system_upgrade.clear_dir(self.tmpdir) self.assertTrue(os.path.isdir(self.tmpdir)) self.assertEqual(os.listdir(self.tmpdir), []) def tearDown(self): shutil.rmtree(self.tmpdir) class CommandTestCaseBase(unittest.TestCase): def setUp(self): self.datadir = tempfile.mkdtemp(prefix="system_upgrade_test_datadir-") self.installroot = tempfile.TemporaryDirectory(prefix="system_upgrade_test_installroot-") system_upgrade.SystemUpgradeCommand.DATADIR = self.datadir self.cli = mock.MagicMock() # the installroot is not strictly necessary for the test, but # releasever detection is accessing host system files without it, and # this fails on permissions in COPR srpm builds (e.g. from rpm-gitoverlay) self.cli.base.conf.installroot = self.installroot.name self.command = system_upgrade.SystemUpgradeCommand(cli=self.cli) self.command.base.conf.cachedir = os.path.join(self.datadir, "cache") self.command.base.conf.destdir = None def tearDown(self): shutil.rmtree(self.datadir) self.installroot.cleanup() class CommandTestCase(CommandTestCaseBase): # self-tests for the command test cases def test_state(self): # initial state: no status self.assertIsNone(self.command.state.download_status) self.assertIsNone(self.command.state.upgrade_status) class CleanCommandTestCase(CommandTestCaseBase): def test_pre_configure_clean(self): with self.command.state as state: state.destdir = "/grape/wine" self.command.pre_configure_clean() self.assertEqual(self.command.base.conf.destdir, "/grape/wine") def test_configure_clean(self): self.cli.demands.root_user = None self.command.configure_clean() self.assertTrue(self.cli.demands.root_user) def test_run_clean(self): with self.command.state as state: state.download_status = "complete" state.upgrade_status = "ready" # make sure the datadir and state info is set up OK self.assertEqual(self.command.state.download_status, "complete") self.assertEqual(self.command.state.upgrade_status, "ready") # run cleanup self.command.run_clean() # state is cleared self.assertIsNone(self.command.state.download_status) self.assertIsNone(self.command.state.upgrade_status) class RebootCheckCommandTestCase(CommandTestCaseBase): def setUp(self): super(RebootCheckCommandTestCase, self).setUp() self.magic_symlink = self.datadir + '/symlink' self.command.magic_symlink = self.magic_symlink def test_pre_configure_reboot(self): with self.command.state as state: state.destdir = "/grape/wine" self.command.pre_configure_reboot() self.assertEqual(self.command.base.conf.destdir, "/grape/wine") def test_configure_reboot(self): self.cli.demands.root_user = None self.command.configure_reboot() self.assertTrue(self.cli.demands.root_user) def check_reboot(self, status='complete', lexists=False, command='system-upgrade', state_command='system-upgrade'): with patch('system_upgrade.os.path.lexists') as lexists_func: self.command.state.state_version = 3 self.command.state.download_status = status self.command.opts = mock.MagicMock() self.command.opts.command = command self.command.state.upgrade_command = state_command lexists_func.return_value = lexists self.command.check_reboot() def test_check_reboot_ok(self): self.check_reboot(status='complete', lexists=False) def test_check_reboot_different_command(self): with self.assertRaises(CliError): self.check_reboot(status='complete', lexists=False, command='system-upgrade', state_command='offline-upgrade') def test_check_reboot_no_download(self): with self.assertRaises(CliError): self.check_reboot(status=None, lexists=False) def test_check_reboot_link_exists(self): with self.assertRaises(CliError): self.check_reboot(status='complete', lexists=True) def test_run_prepare(self): self.command.run_prepare() self.assertEqual(os.readlink(self.magic_symlink), self.datadir) self.assertEqual(self.command.state.upgrade_status, 'ready') @patch('system_upgrade.SystemUpgradeCommand.run_prepare') @patch('system_upgrade.SystemUpgradeCommand.log_status') @patch('system_upgrade.reboot') def test_run_reboot(self, reboot, log_status, run_prepare): self.command.opts = mock.MagicMock() self.command.opts.poweroff_after = False self.command.opts.tid = ["reboot"] self.command.run_reboot() run_prepare.assert_called_once_with() self.assertEqual(system_upgrade.REBOOT_REQUESTED_ID, log_status.call_args[0][1]) self.assertTrue(reboot.called) @patch('system_upgrade.SystemUpgradeCommand.run_prepare') @patch('system_upgrade.SystemUpgradeCommand.log_status') @patch('system_upgrade.reboot') def test_reboot_poweroff_after(self, reboot, log_status, run_prepare): self.command.opts = mock.MagicMock() self.command.opts.tid = ["reboot"] self.command.opts.poweroff_after = True self.command.run_reboot() run_prepare.assert_called_with() self.assertEqual(system_upgrade.REBOOT_REQUESTED_ID, log_status.call_args[0][1]) self.assertTrue(self.command.state.poweroff_after) self.assertTrue(reboot.called) @patch('system_upgrade.SystemUpgradeCommand.run_prepare') @patch('system_upgrade.SystemUpgradeCommand.log_status') @patch('system_upgrade.reboot') def test_reboot_prepare_only(self, reboot, log_status, run_prepare): self.command.opts = mock.MagicMock() self.command.opts.tid = [None] self.command.run_reboot() run_prepare.assert_called_once_with() self.assertFalse(log_status.called) self.assertFalse(reboot.called) class DownloadCommandTestCase(CommandTestCase): def test_pre_configure_download_default(self): self.command.opts = mock.MagicMock() self.command.opts.destdir = None self.command.base.conf.destdir = None self.command.pre_configure_download() self.assertEqual(self.command.base.conf.cachedir, self.datadir) def test_pre_configure_download_destdir(self): self.command.opts = mock.MagicMock() self.command.opts.destdir = self.datadir self.command.pre_configure_download() self.assertEqual(self.command.base.conf.destdir, self.datadir) def test_configure_download(self): self.command.opts = mock.MagicMock() self.command.opts.tid = "download" self.command.configure() self.assertTrue(self.cli.demands.root_user) self.assertTrue(self.cli.demands.resolving) self.assertTrue(self.cli.demands.sack_activation) self.assertTrue(self.cli.demands.available_repos) def test_transaction_download(self): pkg = mock.MagicMock() repo = mock.MagicMock() repo.id = 'test' pkg.name = "kernel" pkg.repo = repo self.cli.base.transaction.install_set = [pkg] self.command.opts = mock.MagicMock() self.command.opts.distro_sync = True self.command.opts.command = "system_upgrade" self.command.opts.repos_ed = [] self.cli.demands.allow_erasing = "allow_erasing" self.command.base.conf.best = True self.command.base.conf.releasever = "35" self.command.base.conf.gpgcheck = True self.command.opts.destdir = self.datadir self.command.base.conf.install_weak_deps = True self.command.base.conf.module_platform_id = '' self.command.pre_configure_download() self.command.transaction_download() with system_upgrade.State(self.command.state.statefile) as state: self.assertEqual(state.state_version, system_upgrade.STATE_VERSION) self.assertEqual(state.download_status, "complete") self.assertEqual(state.distro_sync, True) self.assertEqual(state.destdir, self.datadir) self.assertEqual(state.upgrade_command, "system_upgrade") def test_transaction_download_offline_upgrade(self): pkg = mock.MagicMock() repo = mock.MagicMock() repo.id = 'test' pkg.name = "kernel" pkg.repo = repo self.cli.base.transaction.install_set = [pkg] self.command.opts = mock.MagicMock() self.command.opts.distro_sync = True self.command.opts.command = "offline-upgrade" self.command.opts.repos_ed = [] self.cli.demands.allow_erasing = "allow_erasing" self.command.base.conf.best = True self.command.base.conf.releasever = "35" self.command.base.conf.gpgcheck = True self.command.opts.destdir = self.datadir self.command.base.conf.install_weak_deps = True self.command.base.conf.module_platform_id = '' self.command.pre_configure_download() self.command.transaction_download() with system_upgrade.State(self.command.state.statefile) as state: self.assertEqual(state.download_status, "complete") self.assertEqual(state.distro_sync, False) self.assertEqual(state.destdir, self.datadir) self.assertEqual(state.upgrade_command, "offline-upgrade") class UpgradeCommandTestCase(CommandTestCase): def test_pre_configure_upgrade(self): with self.command.state as state: state.destdir = "/grape/wine" state.target_releasever = "35" self.command.pre_configure_upgrade() self.assertEqual(self.command.base.conf.destdir, "/grape/wine") self.assertEqual(self.command.base.conf.releasever, "35") def test_configure_upgrade(self): # write state like download would have with self.command.state as state: state.download_status = "complete" state.distro_sync = True state.allow_erasing = True state.best = True # okay, now configure upgrade self.command.opts = mock.MagicMock() self.command.opts.tid = "upgrade" self.command.configure() # did we reset the depsolving flags? self.assertTrue(self.command.opts.distro_sync) self.assertTrue(self.cli.demands.allow_erasing) self.assertTrue(self.command.base.conf.best) # are we on autopilot? self.assertTrue(self.command.base.conf.assumeyes) self.assertTrue(self.cli.demands.cacheonly) class LogCommandTestCase(CommandTestCase): def test_configure_log(self): self.command.opts = mock.MagicMock() self.command.opts.tid = "log" self.command.configure() def test_run_log_list(self): self.command.opts = mock.MagicMock() self.command.opts.number = None with patch('system_upgrade.list_logs') as list_logs: self.command.run_log() list_logs.assert_called_once_with() def test_run_log_prev(self): with patch('system_upgrade.show_log') as show_log: self.command.opts = mock.MagicMock() self.command.opts.number = -2 self.command.run_log() show_log.assert_called_once_with(-2)