From 542108b370122295fc0181136c173bb10cff8d5b Mon Sep 17 00:00:00 2001 From: Michael Z Date: Tue, 14 Feb 2023 15:01:33 +0100 Subject: [PATCH] Use native pytest infrastructure So far we've relied on deprecated `importlib` functionality in our tests, as well as our own tmpdir infrastructure. pytest already ships nicer solutions, which this makes use of now. --- tests/__init__.py | 0 tests/test_notify.py | 78 ++++--- tests/test_updater.py | 438 +++++++++++++++++++-------------------- tests/test_updaterapp.py | 47 ++--- tests/test_util.py | 140 ++++++------- 5 files changed, 337 insertions(+), 366 deletions(-) create mode 100644 tests/__init__.py diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_notify.py b/tests/test_notify.py index 2ae3a64..158aaf3 100644 --- a/tests/test_notify.py +++ b/tests/test_notify.py @@ -1,19 +1,11 @@ import datetime -import os import re -from importlib.machinery import SourceFileLoader from unittest import mock import pytest -relpath_notify = "../sdw_notify/Notify.py" -path_to_notify = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_notify) -notify = SourceFileLoader("Notify", path_to_notify).load_module() - -relpath_updater = "../sdw_updater/Updater.py" -path_to_updater = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_updater) -updater = SourceFileLoader("Updater", path_to_updater).load_module() - +from sdw_notify import Notify +from sdw_updater import Updater # Regex for warning log if the last-updated timestamp does not exist (updater # has never run) @@ -41,18 +33,18 @@ BAD_TIMESTAMP_REGEX = r"Data in .* not in the expected format." -@mock.patch("Notify.sdlog.error") -@mock.patch("Notify.sdlog.warning") -@mock.patch("Notify.sdlog.info") -def test_warning_shown_if_updater_never_ran(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_notify.Notify.sdlog.error") +@mock.patch("sdw_notify.Notify.sdlog.warning") +@mock.patch("sdw_notify.Notify.sdlog.info") +def test_warning_shown_if_updater_never_ran(mocked_info, mocked_warning, mocked_error, tmp_path): """ Test whether we're correctly going to show a warning if the updater has never run. """ - # We're going to look for a nonexistent file in an existing tmpdir - with mock.patch("Notify.LAST_UPDATED_FILE", os.path.join(tmpdir, "not-a-file")): + # We're going to look for a nonexistent file in an existing temporary directoryr + with mock.patch("sdw_notify.Notify.LAST_UPDATED_FILE", tmp_path / "not-a-file"): - warning_should_be_shown = notify.is_update_check_necessary() + warning_should_be_shown = Notify.is_update_check_necessary() # No handled errors should occur assert not mocked_error.called @@ -70,13 +62,13 @@ def test_warning_shown_if_updater_never_ran(mocked_info, mocked_warning, mocked_ @pytest.mark.parametrize( "uptime,warning_expected", - [(notify.UPTIME_GRACE_PERIOD + 1, True), (notify.UPTIME_GRACE_PERIOD - 1, False)], + [(Notify.UPTIME_GRACE_PERIOD + 1, True), (Notify.UPTIME_GRACE_PERIOD - 1, False)], ) -@mock.patch("Notify.sdlog.error") -@mock.patch("Notify.sdlog.warning") -@mock.patch("Notify.sdlog.info") +@mock.patch("sdw_notify.Notify.sdlog.error") +@mock.patch("sdw_notify.Notify.sdlog.warning") +@mock.patch("sdw_notify.Notify.sdlog.info") def test_warning_shown_if_warning_threshold_exceeded( - mocked_info, mocked_warning, mocked_error, tmpdir, uptime, warning_expected + mocked_info, mocked_warning, mocked_error, tmp_path, uptime, warning_expected ): """ Primary use case for the notifier: are we showing the warning if the @@ -84,15 +76,15 @@ def test_warning_shown_if_warning_threshold_exceeded( threshold? Expected result varies based on whether system uptime exceeds a grace period (for the user to launch the app on their own). """ - with mock.patch("Notify.LAST_UPDATED_FILE", os.path.join(tmpdir, "sdw-last-updated")): + with mock.patch("sdw_notify.Notify.LAST_UPDATED_FILE", tmp_path / "sdw-last-updated"): # Write a "last successfully updated" date well in the past for check - historic_date = datetime.date(2013, 6, 5).strftime(updater.DATE_FORMAT) - with open(notify.LAST_UPDATED_FILE, "w") as f: + historic_date = datetime.date(2013, 6, 5).strftime(Updater.DATE_FORMAT) + with open(Notify.LAST_UPDATED_FILE, "w") as f: f.write(historic_date) - with mock.patch("Notify.get_uptime_seconds") as mocked_uptime: + with mock.patch("sdw_notify.Notify.get_uptime_seconds") as mocked_uptime: mocked_uptime.return_value = uptime - warning_should_be_shown = notify.is_update_check_necessary() + warning_should_be_shown = Notify.is_update_check_necessary() assert warning_should_be_shown is warning_expected # No handled errors should occur assert not mocked_error.called @@ -108,22 +100,22 @@ def test_warning_shown_if_warning_threshold_exceeded( assert re.search(GRACE_PERIOD_REGEX, info_string) is not None -@mock.patch("Notify.sdlog.error") -@mock.patch("Notify.sdlog.warning") -@mock.patch("Notify.sdlog.info") +@mock.patch("sdw_notify.Notify.sdlog.error") +@mock.patch("sdw_notify.Notify.sdlog.warning") +@mock.patch("sdw_notify.Notify.sdlog.info") def test_warning_not_shown_if_warning_threshold_not_exceeded( - mocked_info, mocked_warning, mocked_error, tmpdir + mocked_info, mocked_warning, mocked_error, tmp_path ): """ Another high priority case: we don't want to warn the user if they've recently run the updater successfully. """ - with mock.patch("Notify.LAST_UPDATED_FILE", os.path.join(tmpdir, "sdw-last-updated")): + with mock.patch("sdw_notify.Notify.LAST_UPDATED_FILE", tmp_path / "sdw-last-updated"): # Write current timestamp into the file - just_now = datetime.datetime.now().strftime(updater.DATE_FORMAT) - with open(notify.LAST_UPDATED_FILE, "w") as f: + just_now = datetime.datetime.now().strftime(Updater.DATE_FORMAT) + with open(Notify.LAST_UPDATED_FILE, "w") as f: f.write(just_now) - warning_should_be_shown = notify.is_update_check_necessary() + warning_should_be_shown = Notify.is_update_check_necessary() assert warning_should_be_shown is False assert not mocked_error.called assert not mocked_warning.called @@ -131,19 +123,19 @@ def test_warning_not_shown_if_warning_threshold_not_exceeded( assert re.search(NO_WARNING_REGEX, info_string) is not None -@mock.patch("Notify.sdlog.error") -@mock.patch("Notify.sdlog.warning") -@mock.patch("Notify.sdlog.info") -def test_corrupt_timestamp_file_handled(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_notify.Notify.sdlog.error") +@mock.patch("sdw_notify.Notify.sdlog.warning") +@mock.patch("sdw_notify.Notify.sdlog.info") +def test_corrupt_timestamp_file_handled(mocked_info, mocked_warning, mocked_error, tmp_path): """ The LAST_UPDATED_FILE must contain a timestamp in a specified format; if it doesn't, we show the warning and log the error. """ - with mock.patch("Notify.LAST_UPDATED_FILE", os.path.join(tmpdir, "sdw-last-updated")): - with open(notify.LAST_UPDATED_FILE, "w") as f: + with mock.patch("sdw_notify.Notify.LAST_UPDATED_FILE", tmp_path / "sdw-last-updated"): + with open(Notify.LAST_UPDATED_FILE, "w") as f: # With apologies to HAL 9000 f.write("daisy, daisy, give me your answer do") - warning_should_be_shown = notify.is_update_check_necessary() + warning_should_be_shown = Notify.is_update_check_necessary() assert warning_should_be_shown is True mocked_error.assert_called_once() error_string = mocked_error.call_args[0][0] @@ -154,6 +146,6 @@ def test_uptime_is_sane(): """ Even in a CI container this should be greater than zero :-) """ - seconds = notify.get_uptime_seconds() + seconds = Notify.get_uptime_seconds() assert isinstance(seconds, float) assert seconds > 0 diff --git a/tests/test_updater.py b/tests/test_updater.py index f94bdb4..0dc1e65 100644 --- a/tests/test_updater.py +++ b/tests/test_updater.py @@ -2,21 +2,13 @@ import os import subprocess from datetime import datetime, timedelta -from importlib.machinery import SourceFileLoader -from tempfile import TemporaryDirectory from unittest import mock from unittest.mock import call import pytest -relpath_updater_script = "../sdw_updater/Updater.py" -path_to_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_updater_script) -updater = SourceFileLoader("Updater", path_to_script).load_module() -from Updater import UpdateStatus # noqa: E402 -from Updater import current_templates # noqa: E402 -from Updater import current_vms # noqa: E402 - -temp_dir = TemporaryDirectory().name +from sdw_updater import Updater +from sdw_updater.Updater import UpdateStatus, current_templates, current_vms debian_based_vms = [ "sd-app", @@ -60,31 +52,31 @@ def test_updater_vms_present(): - assert len(updater.current_vms) == 8 + assert len(Updater.current_vms) == 8 def test_updater_templatevms_present(): - assert len(updater.current_templates) == 4 + assert len(Updater.current_templates) == 4 -@mock.patch("Updater._write_updates_status_flag_to_disk") -@mock.patch("Updater._write_last_updated_flags_to_disk") -@mock.patch("Updater._apply_updates_vm") -@mock.patch("Updater._apply_updates_dom0", return_value=UpdateStatus.UPDATES_OK) -@mock.patch("Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_REQUIRED) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_last_updated_flags_to_disk") +@mock.patch("sdw_updater.Updater._apply_updates_vm") +@mock.patch("sdw_updater.Updater._apply_updates_dom0", return_value=UpdateStatus.UPDATES_OK) +@mock.patch("sdw_updater.Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_REQUIRED) +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_dom0_updates_available( mocked_info, mocked_error, check_dom0, apply_dom0, apply_vm, write_updated, write_status ): - upgrade_generator = updater.apply_updates(["dom0"]) + upgrade_generator = Updater.apply_updates(["dom0"]) results = {} for vm, progress, result in upgrade_generator: results[vm] = result assert progress is not None - assert updater.overall_update_status(results) == UpdateStatus.UPDATES_OK + assert Updater.overall_update_status(results) == UpdateStatus.UPDATES_OK assert not mocked_error.called # Ensure we check for updates, and apply them (with no parameters) check_dom0.assert_called_once_with() @@ -92,24 +84,24 @@ def test_apply_updates_dom0_updates_available( assert not apply_vm.called -@mock.patch("Updater._write_updates_status_flag_to_disk") -@mock.patch("Updater._write_last_updated_flags_to_disk") -@mock.patch("Updater._apply_updates_vm") -@mock.patch("Updater._apply_updates_dom0") -@mock.patch("Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_OK) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_last_updated_flags_to_disk") +@mock.patch("sdw_updater.Updater._apply_updates_vm") +@mock.patch("sdw_updater.Updater._apply_updates_dom0") +@mock.patch("sdw_updater.Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_OK) +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_dom0_no_updates( mocked_info, mocked_error, check_dom0, apply_dom0, apply_vm, write_updated, write_status ): - upgrade_generator = updater.apply_updates(["dom0"]) + upgrade_generator = Updater.apply_updates(["dom0"]) results = {} for vm, progress, result in upgrade_generator: results[vm] = result assert progress is not None - assert updater.overall_update_status(results) == UpdateStatus.UPDATES_OK + assert Updater.overall_update_status(results) == UpdateStatus.UPDATES_OK assert not mocked_error.called # We check for updates, but do not attempt to apply them check_dom0.assert_called_once_with() @@ -117,19 +109,19 @@ def test_apply_updates_dom0_no_updates( assert not apply_vm.called -@mock.patch("Updater._write_updates_status_flag_to_disk") -@mock.patch("Updater._write_last_updated_flags_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_last_updated_flags_to_disk") @mock.patch( - "Updater._apply_updates_vm", + "sdw_updater.Updater._apply_updates_vm", side_effect=[UpdateStatus.UPDATES_OK, UpdateStatus.UPDATES_REQUIRED], ) -@mock.patch("Updater._apply_updates_dom0") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._apply_updates_dom0") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_required( mocked_info, mocked_error, apply_dom0, apply_vm, write_updated, write_status ): - upgrade_generator = updater.apply_updates(["fedora", "sd-app"]) + upgrade_generator = Updater.apply_updates(["fedora", "sd-app"]) results = {} for vm, progress, result in upgrade_generator: @@ -142,23 +134,23 @@ def test_apply_updates_required( assert results == {"fedora": UpdateStatus.UPDATES_OK, "sd-app": UpdateStatus.UPDATES_REQUIRED} - assert updater.overall_update_status(results) == UpdateStatus.UPDATES_REQUIRED + assert Updater.overall_update_status(results) == UpdateStatus.UPDATES_REQUIRED assert not mocked_error.called assert not apply_dom0.called @pytest.mark.parametrize("status", UpdateStatus) -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_write_updates_status_flag_to_disk( - mocked_info, mocked_error, mocked_call, mocked_expand, status + mocked_info, mocked_error, mocked_call, status, tmp_path ): - flag_file_sd_app = updater.FLAG_FILE_STATUS_SD_APP - flag_file_dom0 = updater.get_dom0_path(updater.FLAG_FILE_STATUS_DOM0) + with mock.patch("os.path.expanduser", return_value=tmp_path): + flag_file_sd_app = Updater.FLAG_FILE_STATUS_SD_APP + flag_file_dom0 = Updater.get_dom0_path(Updater.FLAG_FILE_STATUS_DOM0) - updater._write_updates_status_flag_to_disk(status) + Updater._write_updates_status_flag_to_disk(status) mocked_call.assert_called_once_with( ["qvm-run", "sd-app", "echo '{}' > {}".format(status.value, flag_file_sd_app)] @@ -173,47 +165,45 @@ def test_write_updates_status_flag_to_disk( @pytest.mark.parametrize("status", UpdateStatus) -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_write_updates_status_flag_to_disk_failure_app( - mocked_info, mocked_error, mocked_call, mocked_expand, status + mocked_info, mocked_error, mocked_call, status, tmp_path ): - error_calls = [ call("Error writing update status flag to sd-app"), call("Command 'check_call' returned non-zero exit status 1."), ] - updater._write_updates_status_flag_to_disk(status) + with mock.patch("os.path.expanduser", return_value=tmp_path): + Updater._write_updates_status_flag_to_disk(status) mocked_error.assert_has_calls(error_calls) @pytest.mark.parametrize("status", UpdateStatus) @mock.patch("os.path.exists", side_effect=OSError("os_error")) -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_write_updates_status_flag_to_disk_failure_dom0( - mocked_info, mocked_error, mocked_call, mocked_expand, mocked_open, status + mocked_info, mocked_error, mocked_call, mocked_open, status, tmp_path ): - error_calls = [call("Error writing update status flag to dom0"), call("os_error")] - updater._write_updates_status_flag_to_disk(status) + with mock.patch("os.path.expanduser", return_value=tmp_path): + Updater._write_updates_status_flag_to_disk(status) mocked_error.assert_has_calls(error_calls) -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") -def test_write_last_updated_flags_to_disk(mocked_info, mocked_error, mocked_call, mocked_expand): - flag_file_sd_app = updater.FLAG_FILE_LAST_UPDATED_SD_APP - flag_file_dom0 = updater.get_dom0_path(updater.FLAG_FILE_LAST_UPDATED_DOM0) - current_time = str(datetime.now().strftime(updater.DATE_FORMAT)) - - updater._write_last_updated_flags_to_disk() +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") +def test_write_last_updated_flags_to_disk(mocked_info, mocked_error, mocked_call, tmp_path): + flag_file_sd_app = Updater.FLAG_FILE_LAST_UPDATED_SD_APP + with mock.patch("os.path.expanduser", return_value=tmp_path): + flag_file_dom0 = Updater.get_dom0_path(Updater.FLAG_FILE_LAST_UPDATED_DOM0) + current_time = str(datetime.now().strftime(Updater.DATE_FORMAT)) + + Updater._write_last_updated_flags_to_disk() subprocess_command = [ "qvm-run", "sd-app", @@ -226,47 +216,43 @@ def test_write_last_updated_flags_to_disk(mocked_info, mocked_error, mocked_call assert contents == current_time -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") -def test_write_last_updated_flags_to_disk_fails( - mocked_info, mocked_error, mocked_call, mocked_expand -): +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") +def test_write_last_updated_flags_to_disk_fails(mocked_info, mocked_error, mocked_call, tmp_path): error_log = [ call("Error writing last updated flag to sd-app"), call("Command 'check_call' returned non-zero exit status 1."), ] - updater._write_last_updated_flags_to_disk() - + with mock.patch("os.path.expanduser", return_value=tmp_path): + Updater._write_last_updated_flags_to_disk() mocked_error.assert_has_calls(error_log) @mock.patch("os.path.exists", return_value=False) -@mock.patch("os.path.expanduser", return_value=temp_dir) @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_write_last_updated_flags_dom0_folder_creation_fail( - mocked_info, mocked_error, mocked_call, mocked_expand, mocked_path_exists + mocked_info, mocked_error, mocked_call, mocked_path_exists, tmp_path ): error_log = [ call("Error writing last updated flag to sd-app"), call("Command 'check_call' returned non-zero exit status 1."), ] - updater._write_last_updated_flags_to_disk() - + with mock.patch("os.path.expanduser", return_value=tmp_path): + Updater._write_last_updated_flags_to_disk() mocked_error.assert_has_calls(error_log) @mock.patch("subprocess.check_call") -@mock.patch("Updater._write_updates_status_flag_to_disk") -@mock.patch("Updater._write_last_updated_flags_to_disk") -@mock.patch("Updater.shutdown_and_start_vms") -@mock.patch("Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_REQUIRED) -@mock.patch("Updater._apply_updates_vm") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_last_updated_flags_to_disk") +@mock.patch("sdw_updater.Updater.shutdown_and_start_vms") +@mock.patch("sdw_updater.Updater._check_updates_dom0", return_value=UpdateStatus.UPDATES_REQUIRED) +@mock.patch("sdw_updater.Updater._apply_updates_vm") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_dom0_updates_applied( mocked_info, mocked_error, @@ -277,7 +263,7 @@ def test_apply_updates_dom0_updates_applied( write_status, mocked_call, ): - result = updater._apply_updates_dom0() + result = Updater._apply_updates_dom0() assert result == UpdateStatus.REBOOT_REQUIRED mocked_call.assert_called_once_with(["sudo", "qubes-dom0-update", "-y"]) assert not mocked_error.called @@ -285,10 +271,10 @@ def test_apply_updates_dom0_updates_applied( @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_dom0_failure(mocked_info, mocked_error, mocked_call): - result = updater._apply_updates_dom0() + result = Updater._apply_updates_dom0() error_log = [ call("An error has occurred updating dom0. Please contact your administrator."), call("Command 'check_call' returned non-zero exit status 1."), @@ -301,11 +287,11 @@ def test_apply_updates_dom0_failure(mocked_info, mocked_error, mocked_call): @pytest.mark.parametrize("vm", current_templates) @mock.patch("subprocess.check_call", side_effect="0") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_vms(mocked_info, mocked_error, mocked_call, vm): if vm != "dom0": - result = updater._apply_updates_vm(vm) + result = Updater._apply_updates_vm(vm) assert result == UpdateStatus.UPDATES_OK if vm.startswith("fedora") or vm.startswith("whonix"): @@ -321,24 +307,24 @@ def test_apply_updates_vms(mocked_info, mocked_error, mocked_call, vm): @pytest.mark.parametrize("vm", current_templates) @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_updates_vms_fails(mocked_info, mocked_error, mocked_call, vm): error_calls = [ call("An error has occurred updating {}. Please contact your administrator.".format(vm)), call("Command 'check_call' returned non-zero exit status 1."), ] - result = updater._apply_updates_vm(vm) + result = Updater._apply_updates_vm(vm) assert result == UpdateStatus.UPDATES_FAILED mocked_error.assert_has_calls(error_calls) @mock.patch("subprocess.check_call", side_effect=subprocess.CalledProcessError(1, "check_call")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_check_dom0_updates_available(mocked_info, mocked_error, mocked_call): - result = updater._check_updates_dom0() + result = Updater._check_updates_dom0() error_calls = [ call("dom0 updates required, or cannot check for updates"), @@ -349,129 +335,129 @@ def test_check_dom0_updates_available(mocked_info, mocked_error, mocked_call): @mock.patch("subprocess.check_call") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_check_dom0_no_updates_available(mocked_info, mocked_error, mocked_call): - result = updater._check_updates_dom0() + result = Updater._check_updates_dom0() assert not mocked_error.called mocked_info.assert_called_once_with("No updates available for dom0") assert result == UpdateStatus.UPDATES_OK -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_results_updates_ok(mocked_info, mocked_error): - result = updater.overall_update_status(TEST_RESULTS_OK) + result = Updater.overall_update_status(TEST_RESULTS_OK) assert result == UpdateStatus.UPDATES_OK assert not mocked_error.called -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_updates_failed(mocked_info, mocked_error): - result = updater.overall_update_status(TEST_RESULTS_FAILED) + result = Updater.overall_update_status(TEST_RESULTS_FAILED) assert result == UpdateStatus.UPDATES_FAILED assert not mocked_error.called -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_reboot_required(mocked_info, mocked_error): - result = updater.overall_update_status(TEST_RESULTS_REBOOT) + result = Updater.overall_update_status(TEST_RESULTS_REBOOT) assert result == UpdateStatus.REBOOT_REQUIRED assert not mocked_error.called -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_updates_required(mocked_info, mocked_error): - result = updater.overall_update_status(TEST_RESULTS_UPDATES) + result = Updater.overall_update_status(TEST_RESULTS_UPDATES) assert result == UpdateStatus.UPDATES_REQUIRED assert not mocked_error.called -@mock.patch("Updater.last_required_reboot_performed", return_value=True) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.last_required_reboot_performed", return_value=True) +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_reboot_was_done_previously( mocked_info, mocked_error, mocked_reboot_performed ): - result = updater.overall_update_status(TEST_RESULTS_UPDATES) + result = Updater.overall_update_status(TEST_RESULTS_UPDATES) assert result == UpdateStatus.UPDATES_REQUIRED assert not mocked_error.called -@mock.patch("Updater.last_required_reboot_performed", return_value=False) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.last_required_reboot_performed", return_value=False) +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_overall_update_status_reboot_not_done_previously( mocked_info, mocked_error, mocked_reboot_performed ): - result = updater.overall_update_status(TEST_RESULTS_UPDATES) + result = Updater.overall_update_status(TEST_RESULTS_UPDATES) assert result == UpdateStatus.REBOOT_REQUIRED assert not mocked_error.called @pytest.mark.parametrize("vm", current_vms.keys()) @mock.patch("subprocess.check_output") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_safely_shutdown(mocked_info, mocked_error, mocked_output, vm): call_list = [call(["qvm-shutdown", "--wait", "{}".format(vm)], stderr=-1)] - updater._safely_shutdown_vm(vm) + Updater._safely_shutdown_vm(vm) mocked_output.assert_has_calls(call_list) assert not mocked_error.called @pytest.mark.parametrize("vm", current_vms.keys()) @mock.patch("subprocess.check_output", side_effect=["0", "0", "0"]) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_safely_start(mocked_info, mocked_error, mocked_output, vm): call_list = [ call(["qvm-ls", "--running", "--raw-list"], stderr=-1), call(["qvm-start", "--skip-if-running", vm], stderr=-1), ] - updater._safely_start_vm(vm) + Updater._safely_start_vm(vm) mocked_output.assert_has_calls(call_list) assert not mocked_error.called @pytest.mark.parametrize("vm", current_vms.keys()) @mock.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "check_output")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_safely_start_fails(mocked_info, mocked_error, mocked_output, vm): call_list = [ call("Error while starting {}".format(vm)), call("Command 'check_output' returned non-zero exit status 1."), ] - updater._safely_start_vm(vm) + Updater._safely_start_vm(vm) mocked_error.assert_has_calls(call_list) @pytest.mark.parametrize("vm", current_vms.keys()) @mock.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "check_output")) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_safely_shutdown_fails(mocked_info, mocked_error, mocked_call, vm): call_list = [ call("Failed to shut down {}".format(vm)), call("Command 'check_output' returned non-zero exit status 1."), ] - updater._safely_shutdown_vm(vm) + Updater._safely_shutdown_vm(vm) mocked_error.assert_has_calls(call_list) @mock.patch("subprocess.check_output") -@mock.patch("Updater._safely_start_vm") -@mock.patch("Updater._safely_shutdown_vm") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._safely_start_vm") +@mock.patch("sdw_updater.Updater._safely_shutdown_vm") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_shutdown_and_start_vms( mocked_info, mocked_error, mocked_shutdown, mocked_start, mocked_output ): @@ -499,7 +485,7 @@ def test_shutdown_and_start_vms( call("sd-gpg"), call("sd-log"), ] - updater.shutdown_and_start_vms() + Updater.shutdown_and_start_vms() mocked_output.assert_has_calls(sys_vm_kill_calls) mocked_shutdown.assert_has_calls(template_vm_calls + app_vm_calls + sys_vm_shutdown_calls) app_vm_calls_reversed = list(reversed(app_vm_calls)) @@ -508,10 +494,10 @@ def test_shutdown_and_start_vms( @mock.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "check_output")) -@mock.patch("Updater._safely_start_vm") -@mock.patch("Updater._safely_shutdown_vm") -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater._safely_start_vm") +@mock.patch("sdw_updater.Updater._safely_shutdown_vm") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_shutdown_and_start_vms_sysvm_fail( mocked_info, mocked_error, mocked_shutdown, mocked_start, mocked_output ): @@ -546,7 +532,7 @@ def test_shutdown_and_start_vms_sysvm_fail( call("Command 'check_output' returned non-zero exit status 1."), call("None"), ] - updater.shutdown_and_start_vms() + Updater.shutdown_and_start_vms() mocked_output.assert_has_calls(sys_vm_kill_calls) mocked_shutdown.assert_has_calls(template_vm_calls + app_vm_calls) app_vm_calls_reversed = list(reversed(app_vm_calls)) @@ -556,97 +542,97 @@ def test_shutdown_and_start_vms_sysvm_fail( @pytest.mark.parametrize("status", UpdateStatus) @mock.patch("subprocess.check_call") -@mock.patch("os.path.expanduser", return_value=temp_dir) -@mock.patch("Updater.sdlog.error") -def test_read_dom0_update_flag_from_disk( - mocked_error, mocked_expanduser, mocked_subprocess, status -): - updater._write_updates_status_flag_to_disk(status) +@mock.patch("sdw_updater.Updater.sdlog.error") +def test_read_dom0_update_flag_from_disk(mocked_error, mocked_subprocess, status, tmp_path): + with mock.patch("os.path.expanduser", return_value=tmp_path): + Updater._write_updates_status_flag_to_disk(status) - flag_file_dom0 = updater.get_dom0_path(updater.FLAG_FILE_STATUS_DOM0) + flag_file_dom0 = Updater.get_dom0_path(Updater.FLAG_FILE_STATUS_DOM0) - assert os.path.exists(flag_file_dom0) - with open(flag_file_dom0, "r") as f: - contents = json.load(f) - assert contents["status"] == status.value - assert "tmp" in flag_file_dom0 + assert os.path.exists(flag_file_dom0) + with open(flag_file_dom0, "r") as f: + contents = json.load(f) + assert contents["status"] == status.value + assert "tmp" in flag_file_dom0 - assert updater.read_dom0_update_flag_from_disk() == status - json_values = updater.read_dom0_update_flag_from_disk(with_timestamp=True) + assert Updater.read_dom0_update_flag_from_disk() == status + json_values = Updater.read_dom0_update_flag_from_disk(with_timestamp=True) assert json_values["status"] == status.value assert not mocked_error.called +@pytest.mark.parametrize("status", UpdateStatus) @mock.patch("subprocess.check_call") -@mock.patch("os.path.expanduser", return_value=temp_dir) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_read_dom0_update_flag_from_disk_fails( - mocked_info, mocked_error, mocked_expanduser, mocked_subprocess + mocked_info, mocked_error, mocked_subprocess, status, tmp_path ): - - flag_file_dom0 = updater.get_dom0_path(updater.FLAG_FILE_STATUS_DOM0) + with mock.patch("os.path.expanduser", return_value=tmp_path): + flag_file_dom0 = Updater.get_dom0_path(Updater.FLAG_FILE_STATUS_DOM0) + updater_path = tmp_path / ".securedrop_updater" + updater_path.mkdir() with open(flag_file_dom0, "w") as f: f.write("something") info_calls = [call("Cannot read dom0 status flag, assuming first run")] - assert updater.read_dom0_update_flag_from_disk() is None + assert Updater.read_dom0_update_flag_from_disk() is None assert not mocked_error.called mocked_info.assert_has_calls(info_calls) @mock.patch( - "Updater.read_dom0_update_flag_from_disk", + "sdw_updater.Updater.read_dom0_update_flag_from_disk", return_value={ "last_status_update": "1999-09-09 14:12:12", "status": UpdateStatus.REBOOT_REQUIRED.value, }, ) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_last_required_reboot_performed_successful(mocked_info, mocked_error, mocked_read): - result = updater.last_required_reboot_performed() + result = Updater.last_required_reboot_performed() assert result is True assert not mocked_error.called @mock.patch( - "Updater.read_dom0_update_flag_from_disk", + "sdw_updater.Updater.read_dom0_update_flag_from_disk", return_value={ - "last_status_update": str(datetime.now().strftime(updater.DATE_FORMAT)), + "last_status_update": str(datetime.now().strftime(Updater.DATE_FORMAT)), "status": UpdateStatus.REBOOT_REQUIRED.value, }, ) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_last_required_reboot_performed_failed(mocked_info, mocked_error, mocked_read): - result = updater.last_required_reboot_performed() + result = Updater.last_required_reboot_performed() assert result is False assert not mocked_error.called -@mock.patch("Updater.read_dom0_update_flag_from_disk", return_value=None) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk", return_value=None) +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_last_required_reboot_performed_no_file(mocked_info, mocked_error, mocked_read): - result = updater.last_required_reboot_performed() + result = Updater.last_required_reboot_performed() assert result is True assert not mocked_error.called @mock.patch( - "Updater.read_dom0_update_flag_from_disk", + "sdw_updater.Updater.read_dom0_update_flag_from_disk", return_value={ - "last_status_update": str(datetime.now().strftime(updater.DATE_FORMAT)), + "last_status_update": str(datetime.now().strftime(Updater.DATE_FORMAT)), "status": UpdateStatus.UPDATES_OK.value, }, ) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_last_required_reboot_performed_not_required(mocked_info, mocked_error, mocked_read): - result = updater.last_required_reboot_performed() + result = Updater.last_required_reboot_performed() assert result is True assert not mocked_error.called @@ -664,26 +650,26 @@ def test_last_required_reboot_performed_not_required(mocked_info, mocked_error, (UpdateStatus.UPDATES_FAILED, False, False, True), ], ) -@mock.patch("Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") def test_should_run_updater_status_interval_expired( mocked_write, status, rebooted, expect_status_change, expect_updater ): TEST_INTERVAL = 3600 # the updater should always run when checking interval has expired, # regardless of update or reboot status - with mock.patch("Updater.last_required_reboot_performed") as mocked_last: + with mock.patch("sdw_updater.Updater.last_required_reboot_performed") as mocked_last: mocked_last.return_value = rebooted - with mock.patch("Updater.read_dom0_update_flag_from_disk") as mocked_read: + with mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk") as mocked_read: mocked_read.return_value = { "last_status_update": str( (datetime.now() - timedelta(seconds=(TEST_INTERVAL + 10))).strftime( - updater.DATE_FORMAT + Updater.DATE_FORMAT ) ), "status": status.value, } # assuming that the tests won't take an hour to run! - assert expect_updater == updater.should_launch_updater(TEST_INTERVAL) + assert expect_updater == Updater.should_launch_updater(TEST_INTERVAL) assert expect_status_change == mocked_write.called @@ -700,7 +686,7 @@ def test_should_run_updater_status_interval_expired( (UpdateStatus.UPDATES_FAILED, False, False, True), ], ) -@mock.patch("Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") def test_should_run_updater_status_interval_not_expired( mocked_write, status, rebooted, expect_status_change, expect_updater ): @@ -708,62 +694,62 @@ def test_should_run_updater_status_interval_not_expired( # Even if the interval hasn't expired, the updater should only be skipped when: # - the updater status is UPDATESr_OK, or # - the updater status is REBOOT_REQUIRED and the reboot has been performed. - with mock.patch("Updater.last_required_reboot_performed") as mocked_last: + with mock.patch("sdw_updater.Updater.last_required_reboot_performed") as mocked_last: mocked_last.return_value = rebooted - with mock.patch("Updater.read_dom0_update_flag_from_disk") as mocked_read: + with mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk") as mocked_read: mocked_read.return_value = { - "last_status_update": str(datetime.now().strftime(updater.DATE_FORMAT)), + "last_status_update": str(datetime.now().strftime(Updater.DATE_FORMAT)), "status": status.value, } # assuming that the tests won't take an hour to run! - assert expect_updater == updater.should_launch_updater(TEST_INTERVAL) + assert expect_updater == Updater.should_launch_updater(TEST_INTERVAL) assert expect_status_change == mocked_write.called -@mock.patch("Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") def test_should_run_updater_invalid_status(mocked_write): TEST_INTERVAL = 3600 - with mock.patch("Updater.last_required_reboot_performed") as mocked_last: + with mock.patch("sdw_updater.Updater.last_required_reboot_performed") as mocked_last: mocked_last.return_value = True - with mock.patch("Updater.read_dom0_update_flag_from_disk") as mocked_read: + with mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk") as mocked_read: mocked_read.return_value = {} # assuming that the tests won't take an hour to run! - assert updater.should_launch_updater(TEST_INTERVAL) is True + assert Updater.should_launch_updater(TEST_INTERVAL) is True -@mock.patch("Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") def test_should_run_updater_invalid_timestamp(mocked_write): TEST_INTERVAL = 3600 - with mock.patch("Updater.last_required_reboot_performed") as mocked_last: + with mock.patch("sdw_updater.Updater.last_required_reboot_performed") as mocked_last: mocked_last.return_value = True - with mock.patch("Updater.read_dom0_update_flag_from_disk") as mocked_read: + with mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk") as mocked_read: mocked_read.return_value = { "last_status_update": "time to die", "status": UpdateStatus.UPDATES_OK.value, } # assuming that the tests won't take an hour to run! - assert updater.should_launch_updater(TEST_INTERVAL) is True + assert Updater.should_launch_updater(TEST_INTERVAL) is True -@mock.patch("Updater._write_updates_status_flag_to_disk") +@mock.patch("sdw_updater.Updater._write_updates_status_flag_to_disk") def test_should_run_updater_invalid_status_value(mocked_write): TEST_INTERVAL = 3600 - with mock.patch("Updater.last_required_reboot_performed") as mocked_last: + with mock.patch("sdw_updater.Updater.last_required_reboot_performed") as mocked_last: mocked_last.return_value = True - with mock.patch("Updater.read_dom0_update_flag_from_disk") as mocked_read: + with mock.patch("sdw_updater.Updater.read_dom0_update_flag_from_disk") as mocked_read: mocked_read.return_value = { - "last_status_update": str(datetime.now().strftime(updater.DATE_FORMAT)), + "last_status_update": str(datetime.now().strftime(Updater.DATE_FORMAT)), "status": "5", } # assuming that the tests won't take an hour to run! - assert updater.should_launch_updater(TEST_INTERVAL) is True + assert Updater.should_launch_updater(TEST_INTERVAL) is True @mock.patch("subprocess.check_output", side_effect=[b""]) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_dom0_state_success(mocked_info, mocked_error, mocked_subprocess): - updater.apply_dom0_state() + Updater.apply_dom0_state() log_call_list = [call("Applying dom0 state"), call("Dom0 state applied")] mocked_subprocess.assert_called_once_with( ["sudo", "qubesctl", "--show-output", "state.highstate"] @@ -776,10 +762,10 @@ def test_apply_dom0_state_success(mocked_info, mocked_error, mocked_subprocess): "subprocess.check_output", side_effect=[subprocess.CalledProcessError(1, cmd="check_output", output=b"")], ) -@mock.patch("Updater.sdlog.error") -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_apply_dom0_state_failure(mocked_info, mocked_error, mocked_subprocess): - updater.apply_dom0_state() + Updater.apply_dom0_state() log_error_calls = [ call("Failed to apply dom0 state. See updater-detail.log for details."), call("Command 'check_output' returned non-zero exit status 1."), @@ -793,9 +779,9 @@ def test_apply_dom0_state_failure(mocked_info, mocked_error, mocked_subprocess): @mock.patch("os.path.exists", return_value=True) @mock.patch("os.listdir", return_value=["apple", "banana"]) -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_migration_is_required(mocked_info, mocked_listdir, mocked_exists): - assert updater.migration_is_required() is True + assert Updater.migration_is_required() is True assert mocked_info.called_once_with( "Migration is required, will enforce full config during update" ) @@ -803,13 +789,13 @@ def test_migration_is_required(mocked_info, mocked_listdir, mocked_exists): @mock.patch("os.path.exists", return_value=False) @mock.patch("os.listdir", return_value=[]) -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.info") def test_migration_not_required(mocked_info, mocked_listdir, mocked_exists): - assert updater.migration_is_required() is False + assert Updater.migration_is_required() is False assert not mocked_info.called -@mock.patch("Updater.sdlog.info") +@mock.patch("sdw_updater.Updater.sdlog.info") @mock.patch("subprocess.check_output", return_value=b"") @mock.patch("subprocess.check_call") def test_run_full_install(mocked_call, mocked_output, mocked_info): @@ -822,8 +808,8 @@ def test_run_full_install(mocked_call, mocked_output, mocked_info): # subprocess.check_call is mocked, so this directory should never be accessed # by the test. MIGRATION_DIR = "/tmp/potato" # nosec - with mock.patch("Updater.MIGRATION_DIR", MIGRATION_DIR): - result = updater.run_full_install() + with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR): + result = Updater.run_full_install() check_outputs = [call(["sdw-admin", "--apply"])] check_calls = [call(["sudo", "rm", "-rf", MIGRATION_DIR])] assert mocked_output.call_count == 1 @@ -833,7 +819,7 @@ def test_run_full_install(mocked_call, mocked_output, mocked_info): mocked_call.assert_has_calls(check_calls, any_order=False) -@mock.patch("Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.error") @mock.patch( "subprocess.check_output", side_effect=[subprocess.CalledProcessError(1, cmd="check_output", output=b"")], @@ -848,8 +834,8 @@ def test_run_full_install_with_error(mocked_call, mocked_output, mocked_error): And the failure enum is returned """ MIGRATION_DIR = "/tmp/potato" # nosec - with mock.patch("Updater.MIGRATION_DIR", MIGRATION_DIR): - result = updater.run_full_install() + with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR): + result = Updater.run_full_install() calls = [call(["sdw-admin", "--apply"])] assert mocked_output.call_count == 1 assert mocked_call.call_count == 0 @@ -858,7 +844,7 @@ def test_run_full_install_with_error(mocked_call, mocked_output, mocked_error): mocked_output.assert_has_calls(calls, any_order=False) -@mock.patch("Updater.sdlog.error") +@mock.patch("sdw_updater.Updater.sdlog.error") @mock.patch("subprocess.check_output", return_value=b"") @mock.patch( "subprocess.check_call", side_effect=[subprocess.CalledProcessError(1, cmd="check_call")] @@ -872,8 +858,8 @@ def test_run_full_install_with_flag_error(mocked_call, mocked_output, mocked_err And the failure enum is returned """ MIGRATION_DIR = "/tmp/potato" # nosec - with mock.patch("Updater.MIGRATION_DIR", MIGRATION_DIR): - result = updater.run_full_install() + with mock.patch("sdw_updater.Updater.MIGRATION_DIR", MIGRATION_DIR): + result = Updater.run_full_install() check_outputs = [call(["sdw-admin", "--apply"])] check_calls = [call(["sudo", "rm", "-rf", MIGRATION_DIR])] assert mocked_output.call_count == 1 diff --git a/tests/test_updaterapp.py b/tests/test_updaterapp.py index 6b75345..7979a35 100644 --- a/tests/test_updaterapp.py +++ b/tests/test_updaterapp.py @@ -1,19 +1,12 @@ import os import subprocess import unittest -from importlib.machinery import SourceFileLoader from unittest import mock import pytest from PyQt5.QtWidgets import QApplication -relpath_updaterapp_script = "../sdw_updater/UpdaterApp.py" -path_to_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_updaterapp_script) -updater_app = SourceFileLoader("UpdaterApp", path_to_script).load_module() - -relpath_strings_script = "../sdw_updater/strings.py" -path_to_script = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_strings_script) -strings = SourceFileLoader("strings", path_to_script).load_module() +from sdw_updater import UpdaterApp, strings @pytest.fixture(scope="module", autouse=True) @@ -26,42 +19,42 @@ def app(): xvfb.kill() -@mock.patch("UpdaterApp.Util.get_qubes_version", return_value="4.1") -@mock.patch("UpdaterApp.subprocess.check_output", return_value=b"none") +@mock.patch("sdw_util.Util.get_qubes_version", return_value="4.1") +@mock.patch("sdw_updater.UpdaterApp.subprocess.check_output", return_value=b"none") def test_netcheck_no_network_should_fail(mocked_output, mocked_qubes_version): """ When the host machine has no network connectivity Then the error is logged And netcheck returns False """ - assert not updater_app._is_netcheck_successful() + assert not UpdaterApp._is_netcheck_successful() -@mock.patch("Util.get_qubes_version", return_value=None) -@mock.patch("UpdaterApp.logger.error") +@mock.patch("sdw_util.Util.get_qubes_version", return_value=None) +@mock.patch("sdw_updater.UpdaterApp.logger.error") def test_netcheck_no_qubes_should_fail_with_error(mocked_error, mocked_qubes_version): """ When the network connectivity check is run outside of Qubes Then the check should return not succeed And an error should be logged """ - assert not updater_app._is_netcheck_successful() + assert not UpdaterApp._is_netcheck_successful() assert mocked_error.called @mock.patch("subprocess.check_output", return_value=b"full") -@mock.patch("UpdaterApp.Util.get_qubes_version", return_value="4.1") +@mock.patch("sdw_util.Util.get_qubes_version", return_value="4.1") def test_netcheck_should_succeed(mocked_qubes_version, mocked_output): """ When the network connectivity check is run in Qubes And nmcli detects a connection Then the network check should succeed """ - assert updater_app._is_netcheck_successful() + assert UpdaterApp._is_netcheck_successful() -@mock.patch("UpdaterApp.Util.get_qubes_version", return_value="4.1") -@mock.patch("UpdaterApp.logger.error") +@mock.patch("sdw_util.Util.get_qubes_version", return_value="4.1") +@mock.patch("sdw_updater.UpdaterApp.logger.error") @mock.patch("subprocess.check_output", return_value=b"none") def test_updater_app_with_no_connectivity_should_error( mocked_output, mocked_error, mocked_qubes_version @@ -71,15 +64,15 @@ def test_updater_app_with_no_connectivity_should_error( And the network check is unsuccessful Then the network error view should be visible """ - updater_app_dialog = updater_app.UpdaterApp() + updater_app_dialog = UpdaterApp.UpdaterApp() updater_app_dialog._check_network_and_update() assert is_network_fail_view(updater_app_dialog) -@mock.patch("UpdaterApp.Util.get_qubes_version", return_value="4.1") +@mock.patch("sdw_util.Util.get_qubes_version", return_value="4.1") @mock.patch("subprocess.check_output", return_value=b"full") -@mock.patch("UpdaterApp.logger.info") -@mock.patch("UpdaterApp.UpgradeThread") +@mock.patch("sdw_updater.UpdaterApp.logger.info") +@mock.patch("sdw_updater.UpdaterApp.UpgradeThread") def test_updater_app_with_connectivity_should_succeed( mocked_thread, mocked_logger, mocked_output, mocked_qubes_version ): @@ -89,12 +82,12 @@ def test_updater_app_with_connectivity_should_succeed( Then the Preflight Updater should begin to check for updates And the progress view should be visible """ - updater_app_dialog = updater_app.UpdaterApp() + updater_app_dialog = UpdaterApp.UpdaterApp() updater_app_dialog._check_network_and_update() assert is_progress_view(updater_app_dialog) -@mock.patch("UpdaterApp.UpgradeThread") +@mock.patch("sdw_updater.UpdaterApp.UpgradeThread") def test_updater_app_with_override(mocked_thread): """ When the netcheck is overridden (skipped) @@ -102,12 +95,12 @@ def test_updater_app_with_override(mocked_thread): Then `apply updates` should still be called And the progress bar should be visible """ - updater_app_dialog = updater_app.UpdaterApp(should_skip_netcheck=True) + updater_app_dialog = UpdaterApp.UpdaterApp(should_skip_netcheck=True) updater_app_dialog._check_network_and_update() assert is_progress_view(updater_app_dialog) -def is_progress_view(dialog: updater_app.UpdaterApp) -> bool: +def is_progress_view(dialog: UpdaterApp.UpdaterApp) -> bool: """ Helper method to test assumptions about Dialog UI state. """ @@ -120,7 +113,7 @@ def is_progress_view(dialog: updater_app.UpdaterApp) -> bool: ) -def is_network_fail_view(dialog: updater_app.UpdaterApp) -> bool: +def is_network_fail_view(dialog: UpdaterApp.UpdaterApp) -> bool: """ Helper method to test assumptions about Dialog UI state. """ diff --git a/tests/test_util.py b/tests/test_util.py index a27f24c..9e1004c 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -1,11 +1,13 @@ import os import re import subprocess -from importlib.machinery import SourceFileLoader +from pathlib import Path from unittest import mock import pytest +from sdw_util import Util + # Regex for lock conflicts BUSY_LOCK_REGEX = r"Error obtaining lock on '.*'." @@ -15,27 +17,23 @@ CONFLICTING_PROCESS_REGEX = r"Conflicting process .* is currently running." # Fixtures (sample files) for certain tests -FIXTURES_PATH = os.path.join(os.path.dirname(__file__), "fixtures") - -relpath_util = "../sdw_util/Util.py" -path_to_util = os.path.join(os.path.dirname(os.path.abspath(__file__)), relpath_util) -util = SourceFileLoader("Util", path_to_util).load_module() +FIXTURES_PATH = Path(__file__).parent / "fixtures" DEBIAN_VERSION = "bullseye" -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -def test_obtain_lock(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +def test_obtain_lock(mocked_info, mocked_warning, mocked_error, tmp_path): """ Test whether we can successfully obtain an exclusive lock """ - with mock.patch("Util.LOCK_DIRECTORY", tmpdir): + with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): basename = "test-obtain-lock.lock" pid_str = str(os.getpid()) - lh = util.obtain_lock(basename) # noqa: F841 + lh = Util.obtain_lock(basename) # noqa: F841 # No handled exception should occur assert not mocked_error.called # We should be getting a lock handle back @@ -55,10 +53,12 @@ def test_obtain_lock(mocked_info, mocked_warning, mocked_error, tmpdir): assert "POSIX" in lslocks_output -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -def test_cannot_obtain_exclusive_lock_when_busy(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +def test_cannot_obtain_exclusive_lock_when_busy( + mocked_info, mocked_warning, mocked_error, tmp_path +): """ Test whether only a single process can obtan an exclusive lock (basic lockfile behavior). @@ -66,25 +66,25 @@ def test_cannot_obtain_exclusive_lock_when_busy(mocked_info, mocked_warning, moc This is used to prevent multiple preflight updaters or multiple notifiers from being instantiated. """ - with mock.patch("Util.LOCK_DIRECTORY", tmpdir): + with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): basename = "test-exclusive-lock.lock" - lh1 = util.obtain_lock(basename) # noqa: F841 + Util.obtain_lock(basename) # We're running in the same process, so obtaining a lock will succeed. # Instead we're mocking the IOError lockf would raise. with mock.patch("fcntl.lockf", side_effect=IOError()) as mocked_lockf: - lh2 = util.obtain_lock(basename) + lh2 = Util.obtain_lock(basename) mocked_lockf.assert_called_once() assert lh2 is None error_string = mocked_error.call_args[0][0] assert re.search(BUSY_LOCK_REGEX, error_string) is not None -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -def test_cannot_obtain_shared_lock_when_busy(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +def test_cannot_obtain_shared_lock_when_busy(mocked_info, mocked_warning, mocked_error, tmp_path): """ Test whether an exlusive lock on a lock file is successfully detected by means of attempting to obtain a shared, nonexclusive lock on the same @@ -93,44 +93,44 @@ def test_cannot_obtain_shared_lock_when_busy(mocked_info, mocked_warning, mocked In the preflight updater / notifier, this is used to prevent the notification from being displayed when the preflight updater is already open. """ - with mock.patch("Util.LOCK_DIRECTORY", tmpdir): + with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): basename = "test-conflict.lock" - lh = util.obtain_lock(basename) # noqa: F841 + Util.obtain_lock(basename) # We're running in the same process, so obtaining a lock will succeed. # Instead we're mocking the IOError lockf would raise. with mock.patch("fcntl.lockf", side_effect=IOError()) as mocked_lockf: - can_get_lock = util.can_obtain_lock(basename) + can_get_lock = Util.can_obtain_lock(basename) mocked_lockf.assert_called_once() assert can_get_lock is False error_string = mocked_error.call_args[0][0] assert re.search(BUSY_LOCK_REGEX, error_string) is not None -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -def test_no_lockfile_no_problems(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +def test_no_lockfile_no_problems(mocked_info, mocked_warning, mocked_error, tmp_path): """ Test whether our shared lock test succeeds even when there's no lockfile (which means the process has not run recently, or ever, and it's safe to run the potentially conflicting process). """ - with mock.patch("Util.LOCK_DIRECTORY", tmpdir): - lock_result = util.can_obtain_lock("404.lock") + with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): + lock_result = Util.can_obtain_lock("404.lock") assert lock_result is True -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") def test_permission_error_is_handled(mocked_info, mocked_warning, mocked_error): """ Test whether permission errors obtaining a lock are handled correctly """ with mock.patch("builtins.open", side_effect=PermissionError()) as mocked_open: # noqa: F821 - lock = util.obtain_lock("test-open-error.lock") + lock = Util.obtain_lock("test-open-error.lock") assert lock is None mocked_open.assert_called_once() mocked_error.assert_called_once() @@ -138,44 +138,44 @@ def test_permission_error_is_handled(mocked_info, mocked_warning, mocked_error): assert re.search(LOCK_PERMISSION_REGEX, error_string) is not None -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -def test_stale_lockfile_has_no_effect(mocked_info, mocked_warning, mocked_error, tmpdir): +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +def test_stale_lockfile_has_no_effect(mocked_info, mocked_warning, mocked_error, tmp_path): """ Test whether we can get a shared lock when a lockfile exists, but nobody is accessing it. """ - with mock.patch("Util.LOCK_DIRECTORY", tmpdir): + with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): # Because we're not assigning the return value, it will be immediately released basename = "test-stale.lock" - util.obtain_lock(basename) - lock_result = util.can_obtain_lock(basename) + Util.obtain_lock(basename) + lock_result = Util.can_obtain_lock(basename) assert lock_result is True -def test_log(tmpdir): +def test_log(tmp_path): """ Test whether we can successfully write to a log file """ - with mock.patch("Util.LOG_DIRECTORY", tmpdir): + with mock.patch("sdw_util.Util.LOG_DIRECTORY", tmp_path): basename = "test.log" # configure_logging is expected to re-create the directory. - os.rmdir(tmpdir) - util.configure_logging(basename) - util.sdlog.info("info level log entry") - util.sdlog.warning("error level log entry") - util.sdlog.error("error level log entry") - path = os.path.join(tmpdir, basename) - count = len(open(path).readlines()) + os.rmdir(tmp_path) + Util.configure_logging(basename) + Util.sdlog.info("info level log entry") + Util.sdlog.warning("error level log entry") + Util.sdlog.error("error level log entry") + path = tmp_path / basename + count = len(path.open().readlines()) assert count == 3 @pytest.mark.parametrize("return_code,expected_result", [(0, True), (1, False)]) -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") def test_for_conflicting_process( mocked_info, mocked_warning, mocked_error, return_code, expected_result ): @@ -186,7 +186,7 @@ def test_for_conflicting_process( # changes at that level. completed_process = subprocess.CompletedProcess(args=[], returncode=return_code) with mock.patch("subprocess.run", return_value=completed_process) as mocked_run: - running_process = util.is_conflicting_process_running(["cowsay"]) + running_process = Util.is_conflicting_process_running(["cowsay"]) mocked_run.assert_called_once() if expected_result is True: assert running_process is True @@ -206,10 +206,10 @@ def test_for_conflicting_process( ("no-such-file", None), ], ) -@mock.patch("Util.sdlog.error") -@mock.patch("Util.sdlog.warning") -@mock.patch("Util.sdlog.info") -@mock.patch("Util.OS_RELEASE_FILE", os.path.join(FIXTURES_PATH, "os-release-qubes-4.1")) +@mock.patch("sdw_util.Util.sdlog.error") +@mock.patch("sdw_util.Util.sdlog.warning") +@mock.patch("sdw_util.Util.sdlog.info") +@mock.patch("sdw_util.Util.OS_RELEASE_FILE", FIXTURES_PATH / "os-release-qubes-4.1") def test_detect_qubes( mocked_info, mocked_warning, mocked_error, os_release_fixture, version_contains ): @@ -217,8 +217,8 @@ def test_detect_qubes( Test whether we can successfully detect whether we're on Qubes and, if so, what version of Qubes, by parsing /etc/os-release in the expected format. """ - with mock.patch("Util.OS_RELEASE_FILE", os.path.join(FIXTURES_PATH, os_release_fixture)): - qubes_version = util.get_qubes_version() + with mock.patch("sdw_util.Util.OS_RELEASE_FILE", FIXTURES_PATH / os_release_fixture): + qubes_version = Util.get_qubes_version() if version_contains is not None: assert qubes_version is not None assert version_contains in qubes_version @@ -233,9 +233,9 @@ def test_get_logger(): """ test_prefix = "potato" test_module = "salad" - logger = util.get_logger(prefix=test_prefix) + logger = Util.get_logger(prefix=test_prefix) assert logger.name == test_prefix - logger = util.get_logger(prefix=test_prefix, module=test_module) + logger = Util.get_logger(prefix=test_prefix, module=test_module) assert logger.name == "{}.{}".format(test_prefix, test_module) @@ -247,7 +247,7 @@ def test_get_logger(): ("no-such-file", None), ], ) -@mock.patch("Util.OS_RELEASE_FILE", os.path.join(FIXTURES_PATH, "os-release-qubes-4.1")) +@mock.patch("sdw_util.Util.OS_RELEASE_FILE", FIXTURES_PATH / "os-release-qubes-4.1") def test_is_sdapp_halted_yes(os_release_fixture, version_contains): """ When sd-app state is 'Halted' @@ -261,7 +261,7 @@ def test_is_sdapp_halted_yes(os_release_fixture, version_contains): with mock.patch("subprocess.check_output") as patched_subprocess_check: patched_subprocess_check.return_value = output - assert util.is_sdapp_halted() + assert Util.is_sdapp_halted() @pytest.mark.parametrize( @@ -272,7 +272,7 @@ def test_is_sdapp_halted_yes(os_release_fixture, version_contains): ("no-such-file", None), ], ) -@mock.patch("Util.OS_RELEASE_FILE", os.path.join(FIXTURES_PATH, "os-release-qubes-4.1")) +@mock.patch("sdw_util.Util.OS_RELEASE_FILE", FIXTURES_PATH / "os-release-qubes-4.1") def test_is_sdapp_halted_no(os_release_fixture, version_contains): """ When sd-app is not Halted (i.e. Running, Pasued) @@ -286,7 +286,7 @@ def test_is_sdapp_halted_no(os_release_fixture, version_contains): with mock.patch("subprocess.check_output") as patched_subprocess: patched_subprocess.return_value = output - assert not util.is_sdapp_halted() + assert not Util.is_sdapp_halted() @pytest.mark.parametrize( @@ -297,7 +297,7 @@ def test_is_sdapp_halted_no(os_release_fixture, version_contains): ("no-such-file", None), ], ) -@mock.patch("Util.OS_RELEASE_FILE", os.path.join(FIXTURES_PATH, "os-release-qubes-4.1")) +@mock.patch("sdw_util.Util.OS_RELEASE_FILE", FIXTURES_PATH / "os-release-qubes-4.1") @mock.patch("subprocess.check_output", side_effect=subprocess.CalledProcessError(1, "check_output")) def test_is_sdapp_halted_error(patched_subprocess, os_release_fixture, version_contains): """ @@ -306,4 +306,4 @@ def test_is_sdapp_halted_error(patched_subprocess, os_release_fixture, version_c And the method should return False """ - assert not util.is_sdapp_halted() + assert not Util.is_sdapp_halted()