From 59155c60d36cfebb9f8154bc702a4a2983548464 Mon Sep 17 00:00:00 2001 From: Micah Lee Date: Wed, 8 May 2024 10:59:05 -0700 Subject: [PATCH] Fix ruff suggestions --- dom0/remove-tags.py | 5 ++- files/destroy-vm.py | 4 +- files/sdw-admin.py | 10 ++--- files/validate_config.py | 3 +- launcher/tests/test_notify.py | 3 +- launcher/tests/test_updater.py | 10 +++-- launcher/tests/test_util.py | 14 +++---- pyproject.toml | 5 +++ sdw_notify/Notify.py | 29 +++++++------- sdw_notify/NotifyApp.py | 22 +++++------ sdw_updater/Updater.py | 71 +++++++++++++++------------------- sdw_updater/UpdaterApp.py | 3 +- sdw_util/Util.py | 4 +- tests/base.py | 17 +++----- tests/test_app.py | 5 +-- tests/test_dom0_rpm_repo.py | 5 +-- tests/test_gpg.py | 5 +-- tests/test_log_vm.py | 5 +-- tests/test_proxy_vm.py | 5 +-- tests/test_qubes_rpc.py | 3 +- tests/test_qubes_vms.py | 3 +- tests/test_sd_devices.py | 5 +-- tests/test_sd_whonix.py | 5 +-- tests/test_sys_usb.py | 5 +-- tests/test_viewer.py | 5 +-- tests/test_vms_exist.py | 3 +- tests/test_vms_platform.py | 6 +-- 27 files changed, 114 insertions(+), 146 deletions(-) diff --git a/dom0/remove-tags.py b/dom0/remove-tags.py index 37e5dee3..41ff20ca 100755 --- a/dom0/remove-tags.py +++ b/dom0/remove-tags.py @@ -3,6 +3,9 @@ Removes tags used for exempting VMs from default SecureDrop Workstation RPC policies from all VMs (including non-SecureDrop ones). """ + +import sys + import qubesadmin q = qubesadmin.Qubes() @@ -21,7 +24,7 @@ def main(): except Exception as error: print(f"Error removing tag: '{error}'") print("Aborting.") - exit(1) + sys.exit(1) tags_removed = True if tags_removed is False: diff --git a/files/destroy-vm.py b/files/destroy-vm.py index 9057c9b6..e3b829b6 100755 --- a/files/destroy-vm.py +++ b/files/destroy-vm.py @@ -3,6 +3,7 @@ Utility to quickly destroy a Qubes VM managed by the Workstation salt config, for use in repeated builds during development. """ + import argparse import subprocess import sys @@ -38,7 +39,8 @@ def destroy_vm(vm): Destroys a single VM instance. Requires arg to be QubesVM object. """ - assert SDW_DEFAULT_TAG in vm.tags + if SDW_DEFAULT_TAG not in vm.tags: + raise Exception("VM does not have the 'sd-workstation' tag") if vm.is_running(): vm.kill() print(f"Destroying VM '{vm.name}'... ", end="") diff --git a/files/sdw-admin.py b/files/sdw-admin.py index 610de66c..51d5151e 100755 --- a/files/sdw-admin.py +++ b/files/sdw-admin.py @@ -4,6 +4,7 @@ packages only puts the files in place `/srv/salt` but does not apply the state, nor does it handle the config. """ + import argparse import os import subprocess @@ -59,9 +60,7 @@ def parse_args(): action="store_true", help="During uninstall action, don't prompt for confirmation, proceed immediately", ) - args = parser.parse_args() - - return args + return parser.parse_args() def install_pvh_support(): @@ -70,9 +69,9 @@ def install_pvh_support(): TODO: install this via package requirements instead if possible """ try: - subprocess.run(["sudo", "qubes-dom0-update", "-y", "-q", "grub2-xen-pvh"], check=False) + subprocess.check_call(["sudo", "qubes-dom0-update", "-y", "-q", "grub2-xen-pvh"]) except subprocess.CalledProcessError: - raise SDWAdminException("Error installing grub2-xen-pvah: local PVH not available.") + raise SDWAdminException("Error installing grub2-xen-pvh: local PVH not available.") def copy_config(): @@ -141,7 +140,6 @@ def refresh_salt(): def perform_uninstall(keep_template_rpm=False): - try: subprocess.check_call(["sudo", "qubesctl", "state.sls", "sd-clean-default-dispvm"]) print("Destroying all VMs") diff --git a/files/validate_config.py b/files/validate_config.py index 12bdc04e..5cab73fc 100755 --- a/files/validate_config.py +++ b/files/validate_config.py @@ -129,8 +129,7 @@ def confirm_submission_privkey_fingerprint(self): def read_config_file(self): with open(self.config_filepath) as f: - j = json.load(f) - return j + return json.load(f) def validate_existing_size(self): """This method checks for existing private volume size and new diff --git a/launcher/tests/test_notify.py b/launcher/tests/test_notify.py index 158aaf30..f57556a8 100644 --- a/launcher/tests/test_notify.py +++ b/launcher/tests/test_notify.py @@ -43,7 +43,6 @@ def test_warning_shown_if_updater_never_ran(mocked_info, mocked_warning, mocked_ """ # 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() # No handled errors should occur @@ -61,7 +60,7 @@ def test_warning_shown_if_updater_never_ran(mocked_info, mocked_warning, mocked_ @pytest.mark.parametrize( - "uptime,warning_expected", + ("uptime", "warning_expected"), [(Notify.UPTIME_GRACE_PERIOD + 1, True), (Notify.UPTIME_GRACE_PERIOD - 1, False)], ) @mock.patch("sdw_notify.Notify.sdlog.error") diff --git a/launcher/tests/test_updater.py b/launcher/tests/test_updater.py index b5aed6d7..ebb41de7 100644 --- a/launcher/tests/test_updater.py +++ b/launcher/tests/test_updater.py @@ -212,7 +212,9 @@ def test_write_last_updated_flags_to_disk(mocked_info, mocked_error, mocked_call mocked_call.assert_called_once_with(subprocess_command) assert not mocked_error.called assert os.path.exists(flag_file_dom0) - contents = open(flag_file_dom0).read() + with open(flag_file_dom0) as f: + contents = f.read() + assert contents == current_time @@ -294,7 +296,7 @@ def test_apply_updates_vms(mocked_info, mocked_error, mocked_call, vm): result = Updater._apply_updates_vm(vm) assert result == UpdateStatus.UPDATES_OK - if vm.startswith("fedora") or vm.startswith("whonix"): + if vm.startswith(("fedora", "whonix")): expected_salt_state = "update.qubes-vm" else: expected_salt_state = "fpf-apt-repo" @@ -638,7 +640,7 @@ def test_last_required_reboot_performed_not_required(mocked_info, mocked_error, @pytest.mark.parametrize( - "status, rebooted, expect_status_change, expect_updater", + ("status", "rebooted", "expect_status_change", "expect_updater"), [ (UpdateStatus.UPDATES_OK, True, False, True), (UpdateStatus.UPDATES_REQUIRED, True, False, True), @@ -674,7 +676,7 @@ def test_should_run_updater_status_interval_expired( @pytest.mark.parametrize( - "status, rebooted, expect_status_change, expect_updater", + ("status", "rebooted", "expect_status_change", "expect_updater"), [ (UpdateStatus.UPDATES_OK, True, False, False), (UpdateStatus.UPDATES_REQUIRED, True, False, True), diff --git a/launcher/tests/test_util.py b/launcher/tests/test_util.py index edaeb26f..92aa5f91 100644 --- a/launcher/tests/test_util.py +++ b/launcher/tests/test_util.py @@ -30,7 +30,6 @@ def test_obtain_lock(mocked_info, mocked_warning, mocked_error, tmp_path): Test whether we can successfully obtain an exclusive lock """ 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) @@ -67,7 +66,6 @@ def test_cannot_obtain_exclusive_lock_when_busy( from being instantiated. """ with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): - basename = "test-exclusive-lock.lock" Util.obtain_lock(basename) @@ -94,7 +92,6 @@ def test_cannot_obtain_shared_lock_when_busy(mocked_info, mocked_warning, mocked from being displayed when the preflight updater is already open. """ with mock.patch("sdw_util.Util.LOCK_DIRECTORY", tmp_path): - basename = "test-conflict.lock" Util.obtain_lock(basename) @@ -147,7 +144,6 @@ def test_stale_lockfile_has_no_effect(mocked_info, mocked_warning, mocked_error, is accessing it. """ 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) @@ -172,7 +168,7 @@ def test_log(tmp_path): assert count == 3 -@pytest.mark.parametrize("return_code,expected_result", [(0, True), (1, False)]) +@pytest.mark.parametrize(("return_code", "expected_result"), [(0, True), (1, False)]) @mock.patch("sdw_util.Util.sdlog.error") @mock.patch("sdw_util.Util.sdlog.warning") @mock.patch("sdw_util.Util.sdlog.info") @@ -199,7 +195,7 @@ def test_for_conflicting_process( @pytest.mark.parametrize( - "os_release_fixture,version_contains", + ("os_release_fixture", "version_contains"), [ ("os-release-qubes-4.1", "4.1"), ("os-release-ubuntu", None), @@ -240,7 +236,7 @@ def test_get_logger(): @pytest.mark.parametrize( - "os_release_fixture,version_contains", + ("os_release_fixture", "version_contains"), [ ("os-release-qubes-4.1", "4.1"), ("os-release-ubuntu", None), @@ -265,7 +261,7 @@ def test_is_sdapp_halted_yes(os_release_fixture, version_contains): @pytest.mark.parametrize( - "os_release_fixture,version_contains", + ("os_release_fixture", "version_contains"), [ ("os-release-qubes-4.1", "4.1"), ("os-release-ubuntu", None), @@ -290,7 +286,7 @@ def test_is_sdapp_halted_no(os_release_fixture, version_contains): @pytest.mark.parametrize( - "os_release_fixture,version_contains", + ("os_release_fixture", "version_contains"), [ ("os-release-qubes-4.1", "4.1"), ("os-release-ubuntu", None), diff --git a/pyproject.toml b/pyproject.toml index 884d4d25..9fca55a7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -56,12 +56,15 @@ select = [ ] ignore = [ # code complexity checks that we fail + "PLR0911", "PLR0913", "PLR0915", # magic-value-comparison, too many violations for now "PLR2004", # hardcoded passwords, false positive "S105", + # it's fine to use /tmp in dom0, since it's not a multiuser environment + "S108", # flags every instance of subprocess "S603", # we trust $PATH isn't hijacked @@ -77,6 +80,8 @@ ignore = [ "**/tests/**.py" = [ # Use of `assert` detected "S101", + # Tests use /tmp + "S108", # Use a regular `assert` instead of unittest-style `assertEqual` "PT009", ] diff --git a/sdw_notify/Notify.py b/sdw_notify/Notify.py index 7dee5c03..81c55de9 100644 --- a/sdw_notify/Notify.py +++ b/sdw_notify/Notify.py @@ -78,7 +78,7 @@ def is_update_check_necessary(): "Updater may never have run. Showing security warning." ) return True - elif updated_seconds_ago > WARNING_THRESHOLD: + if updated_seconds_ago > WARNING_THRESHOLD: if uptime_seconds > UPTIME_GRACE_PERIOD: sdlog.warning( f"Last successful update ({updated_hours_ago:.1f} hours ago) is above " @@ -87,25 +87,24 @@ def is_update_check_necessary(): "Showing security warning." ) return True - else: - sdlog.info( - f"Last successful update ({updated_hours_ago:.1f} hours ago) is above " - f"warning threshold ({warning_threshold_hours:.1f} hours). Uptime grace period " - f"of {grace_period_hours:.1f} hours has not elapsed yet (uptime: {uptime_hours:.1f} " - "hours). Exiting without warning." - ) - return False - else: + sdlog.info( - f"Last successful update ({updated_hours_ago:.1f} hours ago) " - f"is below the warning threshold ({warning_threshold_hours:.1f} hours). " - "Exiting without warning." + f"Last successful update ({updated_hours_ago:.1f} hours ago) is above " + f"warning threshold ({warning_threshold_hours:.1f} hours). Uptime grace period " + f"of {grace_period_hours:.1f} hours has not elapsed yet (uptime: {uptime_hours:.1f} " + "hours). Exiting without warning." ) return False + sdlog.info( + f"Last successful update ({updated_hours_ago:.1f} hours ago) " + f"is below the warning threshold ({warning_threshold_hours:.1f} hours). " + "Exiting without warning." + ) + return False + def get_uptime_seconds(): # Obtain current uptime with open("/proc/uptime") as f: - uptime_seconds = float(f.readline().split()[0]) - return uptime_seconds + return float(f.readline().split()[0]) diff --git a/sdw_notify/NotifyApp.py b/sdw_notify/NotifyApp.py index 191b9c88..18bee6d5 100644 --- a/sdw_notify/NotifyApp.py +++ b/sdw_notify/NotifyApp.py @@ -61,18 +61,16 @@ def run(self) -> NotifyStatus: result = self.exec_() if result == QMessageBox.Ok: - sdlog.info( - f"NotfyDialog returned {result}, user has opted to check for updates" - ) + sdlog.info(f"NotfyDialog returned {result}, user has opted to check for updates") return NotifyStatus.CHECK_UPDATES - elif result == QMessageBox.No: + if result == QMessageBox.No: sdlog.info(f"NotfyDialog returned {result}, user has opted to defer updates") return NotifyStatus.DEFER_UPDATES - else: - # Should not occur, as this dialog which can only return - # one of two states. - sdlog.error( - f"NotfyDialog returned unexpected value {result}; consult " - "QMessageBox API for more information" - ) - return NotifyStatus.ERROR_UNKNOWN + + # Should not occur, as this dialog which can only return + # one of two states. + sdlog.error( + f"NotfyDialog returned unexpected value {result}; consult " + "QMessageBox API for more information" + ) + return NotifyStatus.ERROR_UNKNOWN diff --git a/sdw_updater/Updater.py b/sdw_updater/Updater.py index f5252e28..51cc165c 100644 --- a/sdw_updater/Updater.py +++ b/sdw_updater/Updater.py @@ -71,9 +71,7 @@ def run_full_install(): sdlog.error(f"Failed to apply full system state. Please review {DETAIL_LOG_FILE}.") sdlog.error(str(e)) clean_output = Util.strip_ansi_colors(e.output.decode("utf-8").strip()) - detail_log.error( - f"Output from failed command: {apply_cmd_for_log}\n{clean_output}" - ) + detail_log.error(f"Output from failed command: {apply_cmd_for_log}\n{clean_output}") return UpdateStatus.UPDATES_FAILED clean_output = Util.strip_ansi_colors(output.decode("utf-8").strip()) @@ -97,10 +95,9 @@ def migration_is_required(): Check whether a full run of the Salt config via sdw-admin is required. """ result = False - if os.path.exists(MIGRATION_DIR): - if len(os.listdir(MIGRATION_DIR)) > 0: - sdlog.info("Migration is required, will enforce full config during update") - result = True + if os.path.exists(MIGRATION_DIR) and len(os.listdir(MIGRATION_DIR)) > 0: + sdlog.info("Migration is required, will enforce full config during update") + result = True return result @@ -113,7 +110,8 @@ def apply_updates(vms=current_templates, progress_start=15, progress_end=75): """ sdlog.info(f"Applying all updates to VMs: {vms}") # Figure out how much each completed VM should bump the progress bar. - assert progress_end > progress_start + if progress_end <= progress_start: + raise Exception("Invalid progress range") progress_step = (progress_end - progress_start) // len(vms) progress_current = progress_start @@ -185,19 +183,14 @@ def _apply_updates_vm(vm): sdlog.info(f"Updating {vm}") # We run custom Salt logic for our own Debian-based TemplateVMs - if vm.startswith("fedora") or vm.startswith("whonix"): - salt_state = "update.qubes-vm" - else: - salt_state = "fpf-apt-repo" + salt_state = "update.qubes-vm" if vm.startswith(("fedora", "whonix")) else "fpf-apt-repo" try: subprocess.check_call( ["sudo", "qubesctl", "--skip-dom0", "--targets", vm, "state.sls", salt_state] ) except subprocess.CalledProcessError as e: - sdlog.error( - f"An error has occurred updating {vm}. Please contact your administrator." - ) + sdlog.error(f"An error has occurred updating {vm}. Please contact your administrator.") sdlog.error(str(e)) return UpdateStatus.UPDATES_FAILED sdlog.info(f"{vm} update successful") @@ -290,13 +283,13 @@ def last_required_reboot_performed(): # the updater, system was not rebooted after previous run if boot_time < reboot_time: return False + # system was rebooted after flag was written to disk - else: - return True - # previous run did not require reboot - else: return True + # previous run did not require reboot + return True + def _get_uptime(): """ @@ -310,9 +303,7 @@ def _get_uptime(): uptime_minutes = (uptime % 3600) // 60 uptime_seconds = uptime % 60 - delta = timedelta(hours=uptime_hours, minutes=uptime_minutes, seconds=uptime_seconds) - - return delta + return timedelta(hours=uptime_hours, minutes=uptime_minutes, seconds=uptime_seconds) def read_dom0_update_flag_from_disk(with_timestamp=False): @@ -330,8 +321,8 @@ def read_dom0_update_flag_from_disk(with_timestamp=False): if int(contents["status"]) == int(status.value): if with_timestamp: return contents - else: - return status + + return status except Exception: sdlog.info("Cannot read dom0 status flag, assuming first run") return None @@ -361,12 +352,12 @@ def overall_update_status(results): if updates_failed: return UpdateStatus.UPDATES_FAILED - elif reboot_required: + if reboot_required: return UpdateStatus.REBOOT_REQUIRED - elif updates_required: + if updates_required: return UpdateStatus.UPDATES_REQUIRED - else: - return UpdateStatus.UPDATES_OK + + return UpdateStatus.UPDATES_OK def apply_dom0_state(): @@ -475,30 +466,30 @@ def should_launch_updater(interval): if _interval_expired(interval, status): sdlog.info("Update interval expired: launching updater.") return True - elif status["status"] == UpdateStatus.UPDATES_OK.value: + if status["status"] == UpdateStatus.UPDATES_OK.value: sdlog.info("Updates OK and interval not expired, launching client.") return False - elif status["status"] == UpdateStatus.REBOOT_REQUIRED.value: + if status["status"] == UpdateStatus.REBOOT_REQUIRED.value: if last_required_reboot_performed(): sdlog.info("Required reboot performed, updating status and launching client.") _write_updates_status_flag_to_disk(UpdateStatus.UPDATES_OK) return False - else: - sdlog.info("Required reboot pending, launching updater") - return True - elif status["status"] == UpdateStatus.UPDATES_REQUIRED.value: + + sdlog.info("Required reboot pending, launching updater") + return True + if status["status"] == UpdateStatus.UPDATES_REQUIRED.value: sdlog.info("Updates are required, launching updater.") return True - elif status["status"] == UpdateStatus.UPDATES_FAILED.value: + if status["status"] == UpdateStatus.UPDATES_FAILED.value: sdlog.info("Preceding update failed, launching updater.") return True - else: - sdlog.info("Update status is unknown, launching updater.") - return True - else: - sdlog.info("Update status not available, launching updater.") + + sdlog.info("Update status is unknown, launching updater.") return True + sdlog.info("Update status not available, launching updater.") + return True + def _valid_status(status): """ diff --git a/sdw_updater/UpdaterApp.py b/sdw_updater/UpdaterApp.py index a7af5855..9b48db23 100644 --- a/sdw_updater/UpdaterApp.py +++ b/sdw_updater/UpdaterApp.py @@ -27,7 +27,7 @@ def launch_securedrop_client(): class UpdaterApp(QDialog, Ui_UpdaterDialog): def __init__(self, should_skip_netcheck: bool = False, parent=None): - super(UpdaterApp, self).__init__(parent) + super().__init__(parent) self.progress = 0 self._skip_netcheck = should_skip_netcheck @@ -222,7 +222,6 @@ def __init__(self): QThread.__init__(self) def run(self): - # Update dom0 first, then apply dom0 state. If full state run # is required, the dom0 state will drop a flag. self.progress_signal.emit(5) diff --git a/sdw_util/Util.py b/sdw_util/Util.py index c0928e51..b5867690 100644 --- a/sdw_util/Util.py +++ b/sdw_util/Util.py @@ -151,8 +151,8 @@ def get_qubes_version(): def get_logger(prefix=SD_LOGGER_PREFIX, module=None): if module is None: return logging.getLogger(prefix) - else: - return logging.getLogger(prefix + "." + module) + + return logging.getLogger(prefix + "." + module) def strip_ansi_colors(str): diff --git a/tests/base.py b/tests/base.py index 0f4f8773..d5cf772d 100644 --- a/tests/base.py +++ b/tests/base.py @@ -91,18 +91,15 @@ def _run(self, cmd, user=""): if user: full_cmd += ["-u", user] full_cmd += [self.vm_name, cmd] - contents = subprocess.check_output(full_cmd).decode("utf-8").strip() - return contents + return subprocess.check_output(full_cmd).decode("utf-8").strip() def _get_file_contents(self, path): cmd = ["qvm-run", "-p", self.vm_name, f"sudo /bin/cat {path}"] - contents = subprocess.check_output(cmd).decode("utf-8") - return contents + return subprocess.check_output(cmd).decode("utf-8") def _get_symlink_location(self, path): cmd = ["qvm-run", "-p", self.vm_name, f"/usr/bin/readlink -f {path}"] - contents = subprocess.check_output(cmd).decode("utf-8").strip() - return contents + return subprocess.check_output(cmd).decode("utf-8").strip() def _package_is_installed(self, pkg): """ @@ -111,9 +108,7 @@ def _package_is_installed(self, pkg): # dpkg --verify will exit non-zero for a non-installed pkg, # catch that and return False try: - subprocess.check_call( - ["qvm-run", "-a", "-q", self.vm_name, f"dpkg --verify {pkg}"] - ) + subprocess.check_call(["qvm-run", "-a", "-q", self.vm_name, f"dpkg --verify {pkg}"]) except subprocess.CalledProcessError: return False @@ -145,9 +140,7 @@ def _fileExists(self, remote_path): # ls will return non-zero and an exception will be thrown if the file # does not exist, so we return false in that case. try: - subprocess.check_call( - ["qvm-run", "-a", "-q", self.vm_name, f"ls {remote_path}"] - ) + subprocess.check_call(["qvm-run", "-a", "-q", self.vm_name, f"ls {remote_path}"]) except subprocess.CalledProcessError: return False diff --git a/tests/test_app.py b/tests/test_app.py index e8049816..14563bdc 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -7,7 +7,7 @@ class SD_App_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-app" - super(SD_App_Tests, self).setUp() + super().setUp() self.expected_config_keys = {"QUBES_GPG_DOMAIN", "SD_SUBMISSION_KEY_FPR"} def test_open_in_dvm_desktop(self): @@ -58,5 +58,4 @@ def test_logging_configured(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_App_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_App_Tests) diff --git a/tests/test_dom0_rpm_repo.py b/tests/test_dom0_rpm_repo.py index 372767f1..5b5e7a9c 100644 --- a/tests/test_dom0_rpm_repo.py +++ b/tests/test_dom0_rpm_repo.py @@ -6,7 +6,6 @@ class SD_Dom0_Rpm_Repo_Tests(unittest.TestCase): - pubkey_wanted = "" yum_repo_url = "" pubkey_actual = "/etc/pki/rpm-gpg/RPM-GPG-KEY-securedrop-workstation" @@ -31,7 +30,6 @@ def setUp(self): self.yum_repo_url = self.yum_repo_url_test def test_rpm_repo_public_key(self): - with open(self.pubkey_actual) as f: pubkey_actual_contents = f.readlines() @@ -57,5 +55,4 @@ def test_rpm_repo_config(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Dom0_Rpm_Repo_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Dom0_Rpm_Repo_Tests) diff --git a/tests/test_gpg.py b/tests/test_gpg.py index 3e179f42..64ac8e8e 100644 --- a/tests/test_gpg.py +++ b/tests/test_gpg.py @@ -9,7 +9,7 @@ class SD_GPG_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-gpg" - super(SD_GPG_Tests, self).setUp() + super().setUp() def getLocalFingerprint(self): """ @@ -82,5 +82,4 @@ def test_logging_disabled(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_GPG_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_GPG_Tests) diff --git a/tests/test_log_vm.py b/tests/test_log_vm.py index 6f00ef2c..6b7c1a2b 100644 --- a/tests/test_log_vm.py +++ b/tests/test_log_vm.py @@ -6,7 +6,7 @@ class SD_Log_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-log" - super(SD_Log_Tests, self).setUp() + super().setUp() def test_sd_log_package_installed(self): self.assertTrue(self._package_is_installed("securedrop-log")) @@ -50,5 +50,4 @@ def test_log_dirs_properly_named(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Log_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Log_Tests) diff --git a/tests/test_proxy_vm.py b/tests/test_proxy_vm.py index e5dfc23c..81b24daf 100644 --- a/tests/test_proxy_vm.py +++ b/tests/test_proxy_vm.py @@ -6,7 +6,7 @@ class SD_Proxy_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-proxy" - super(SD_Proxy_Tests, self).setUp() + super().setUp() self.expected_config_keys = {"SD_PROXY_ORIGIN"} def test_do_not_open_here(self): @@ -52,5 +52,4 @@ def test_mailcap_hardened(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Proxy_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Proxy_Tests) diff --git a/tests/test_qubes_rpc.py b/tests/test_qubes_rpc.py index e87e57f3..afe44f56 100644 --- a/tests/test_qubes_rpc.py +++ b/tests/test_qubes_rpc.py @@ -79,5 +79,4 @@ def test_qubesgpg_from_other_to_sdgpg_denied(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Qubes_Rpc_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Qubes_Rpc_Tests) diff --git a/tests/test_qubes_vms.py b/tests/test_qubes_vms.py index 8b176cfb..f42d486c 100644 --- a/tests/test_qubes_vms.py +++ b/tests/test_qubes_vms.py @@ -54,5 +54,4 @@ def test_current_whonix_vms(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Qubes_VM_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Qubes_VM_Tests) diff --git a/tests/test_sd_devices.py b/tests/test_sd_devices.py index d997870b..d1acee60 100644 --- a/tests/test_sd_devices.py +++ b/tests/test_sd_devices.py @@ -7,7 +7,7 @@ class SD_Devices_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-devices-dvm" - super(SD_Devices_Tests, self).setUp() + super().setUp() def test_files_are_properly_copied(self): self.assertTrue(self._fileExists("/usr/bin/send-to-usb")) @@ -51,5 +51,4 @@ def test_open_in_dvm_desktop(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Devices_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Devices_Tests) diff --git a/tests/test_sd_whonix.py b/tests/test_sd_whonix.py index 601978e3..d0b206a8 100644 --- a/tests/test_sd_whonix.py +++ b/tests/test_sd_whonix.py @@ -9,7 +9,7 @@ class SD_Whonix_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-whonix" self.whonix_apt_list = "/etc/apt/sources.list.d/derivative.list" - super(SD_Whonix_Tests, self).setUp() + super().setUp() self.expected_config_keys = {"SD_HIDSERV_HOSTNAME", "SD_HIDSERV_KEY"} def test_accept_sd_xfer_extracted_file(self): @@ -66,5 +66,4 @@ def test_whonix_torrc(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Whonix_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Whonix_Tests) diff --git a/tests/test_sys_usb.py b/tests/test_sys_usb.py index 2d477bea..d2065004 100644 --- a/tests/test_sys_usb.py +++ b/tests/test_sys_usb.py @@ -6,7 +6,7 @@ class SD_SysUSB_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sys-usb" - super(SD_SysUSB_Tests, self).setUp() + super().setUp() def test_files_are_properly_copied(self): self.assertTrue(self._fileExists("/etc/udev/rules.d/99-sd-devices.rules")) @@ -14,5 +14,4 @@ def test_files_are_properly_copied(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_SysUSB_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_SysUSB_Tests) diff --git a/tests/test_viewer.py b/tests/test_viewer.py index ce307879..7fadf734 100644 --- a/tests/test_viewer.py +++ b/tests/test_viewer.py @@ -7,7 +7,7 @@ class SD_Viewer_Tests(SD_VM_Local_Test): def setUp(self): self.vm_name = "sd-viewer" - super(SD_Viewer_Tests, self).setUp() + super().setUp() def test_sd_viewer_metapackage_installed(self): self.assertTrue(self._package_is_installed("securedrop-workstation-viewer")) @@ -54,5 +54,4 @@ def test_mimetypes_symlink(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_Viewer_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_Viewer_Tests) diff --git a/tests/test_vms_exist.py b/tests/test_vms_exist.py index 6ceff688..3880843c 100644 --- a/tests/test_vms_exist.py +++ b/tests/test_vms_exist.py @@ -191,5 +191,4 @@ def sd_large_template(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_VM_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_VM_Tests) diff --git a/tests/test_vms_platform.py b/tests/test_vms_platform.py index 984732b0..1c070362 100644 --- a/tests/test_vms_platform.py +++ b/tests/test_vms_platform.py @@ -36,8 +36,7 @@ def _get_platform_info(self, vm): # let's maintain the default config and retrieve the value elsewise. cmd = "perl -nE '/^PRETTY_NAME=\"(.*)\"$/ and say $1' /etc/os-release" stdout, stderr = vm.run(cmd) - platform = stdout.decode("utf-8").rstrip("\n") - return platform + return stdout.decode("utf-8").rstrip("\n") def _validate_vm_platform(self, vm): """ @@ -191,5 +190,4 @@ def test_all_sd_vm_apt_sources(self): def load_tests(loader, tests, pattern): - suite = unittest.TestLoader().loadTestsFromTestCase(SD_VM_Platform_Tests) - return suite + return unittest.TestLoader().loadTestsFromTestCase(SD_VM_Platform_Tests)