Merge pull request #1487 from freedomofpress/1338-pytest-conversion

 Partial pytest conversion to centralize list of present VMs

Author: legoktm

tests/base.py

@@ -10,6 +10,9 @@
 SD_TEMPLATE_LARGE = f"sd-large-{DEBIAN_VERSION}-template"
 SD_TEMPLATE_SMALL = f"sd-small-{DEBIAN_VERSION}-template"
 
+SD_TAG = "sd-workstation"  # Tag identifying SecureDrop Workstation-managed VMs
+
+# Expectations regarding VMs' existence and versions
 SD_VMS = ["sd-gpg", "sd-log", "sd-proxy", "sd-app", "sd-viewer", "sd-devices"]
 SD_DVM_TEMPLATES = ["sd-devices-dvm", "sd-proxy-dvm"]
 SD_TEMPLATES = [SD_TEMPLATE_BASE, SD_TEMPLATE_LARGE, SD_TEMPLATE_SMALL]

tests/conftest.py

@@ -2,6 +2,9 @@
 from pathlib import Path
 
 import pytest
+from qubesadmin import Qubes
+
+from tests.base import SD_TAG
 
 PROJ_ROOT = Path(__file__).parent.parent
 
@@ -11,3 +14,24 @@ def dom0_config():
     """Make the dom0 "config.json" available to tests."""
     with open(PROJ_ROOT / "config.json") as config_file:
         return json.load(config_file)
+
+
+@pytest.fixture
+def all_vms():
+    """Obtain all qubes present in the system"""
+    return Qubes().domains
+
+
+@pytest.fixture
+def sdw_tagged_vms(all_vms):
+    """Obtain all SecureDrop Workstation-exclusive qubes"""
+    return [vm for vm in all_vms if SD_TAG in vm.tags]
+
+
+@pytest.fixture
+def config():
+    with open("config.json") as c:
+        config = json.load(c)
+    if "environment" not in config:
+        config["environment"] = "dev"
+    return config

tests/test_log_vm.py

@@ -44,16 +44,14 @@ def test_redis_service_running(qube):
     assert qube.service_is_active("redis")
 
 
-def test_logs_are_flowing(qube):
+def test_logs_are_flowing(qube, sdw_tagged_vms):
     """
     To test that logs work, we run a unique command in each VM we care
     about that gets logged, and then check for that string in the logs.
     """
     # Random string, to avoid collisions with other test runs
     token = "".join(secrets.choice(string.ascii_uppercase) for _ in range(10))
 
-    # All @tag:sd-workstation VMs
-    all_vms = [vm.name for vm in qube.app.domains if "sd-workstation" in vm.tags]
     # base template doesn't have sd-log configured
     # TODO: test a sd-viewer based dispVM
     skip = [f"sd-base-{CURRENT_DEBIAN_VERSION}-template", "sd-viewer"]
@@ -63,17 +61,17 @@ def test_logs_are_flowing(qube):
     # We first run the command in each VM, and then do a second loop to
     # look for the token in the log entry, so there's enough time for the
     # log entry to get written.
-    for vm_name in all_vms:
-        if vm_name in skip:
+    for vm in sdw_tagged_vms:
+        if vm.name in skip:
             continue
         # The sudo call will make it into syslog
-        subprocess.check_call(["qvm-run", vm_name, f"sudo echo {token}"])
+        subprocess.check_call(["qvm-run", vm.name, f"sudo echo {token}"])
 
-    for vm_name in all_vms:
-        if vm_name in skip:
+    for vm in sdw_tagged_vms:
+        if vm.name in skip:
             continue
-        syslog = f"/home/user/QubesIncomingLogs/{vm_name}/syslog.log"
-        if vm_name in no_log_vms:
+        syslog = f"/home/user/QubesIncomingLogs/{vm.name}/syslog.log"
+        if vm.name in no_log_vms:
             assert not qube.fileExists(syslog)
         else:
             assert token in qube.get_file_contents(syslog)

tests/test_qubes_rpc.py

@@ -1,68 +1,58 @@
 import functools
 import os
 import subprocess
-import unittest
-
-from qubesadmin import Qubes
-
-
-class SD_Qubes_Rpc_Tests(unittest.TestCase):
-    @classmethod
-    def setUpClass(cls):
-        cls.all_vms = set()
-        cls.vms_by_tag = {}
-
-        app = Qubes()
-        cls.all_vms = {vm for vm in app.domains if vm.name != "dom0"}
-        cls.sdw_tagged_vms = {vm for vm in app.domains if "sd-workstation" in vm.tags}
-
-    @functools.cache
-    def _qrexec_policy_graph(self, service):
-        cmd = ["qrexec-policy-graph", "--service", service]
-        p = subprocess.run(cmd, capture_output=True, text=True, check=False)
-        return p.stdout
-
-    def _policy_exists(self, source, target, service):
-        service_policy_graph = self._qrexec_policy_graph(service)
-        policy_str = f'"{source}" -> "{target}" [label="{service}"'
-        return policy_str in service_policy_graph
-
-    def test_policy_files_exist(self):
-        """verify the policies are installed"""
-        assert os.path.exists("/etc/qubes/policy.d/31-securedrop-workstation.policy")
-        assert os.path.exists("/etc/qubes/policy.d/32-securedrop-workstation.policy")
-
-    # securedrop.Log from @tag:sd-workstation to sd-log should be allowed
-    def test_sdlog_from_sdw_to_sdlog_allowed(self):
-        for vm in self.sdw_tagged_vms:
-            if vm != "sd-log":
-                assert self._policy_exists(vm, "sd-log", "securedrop.Log")
-
-    # securedrop.Log from anything else to sd-log should be denied
-    def test_sdlog_from_other_to_sdlog_denied(self):
-        non_sd_workstation_vms = self.all_vms.difference(self.sdw_tagged_vms)
-        for vm in non_sd_workstation_vms:
-            if vm != "sd-log":
-                assert not self._policy_exists(vm, "sd-log", "securedrop.Log")
-
-    # securedrop.Proxy from sd-app to sd-proxy should be allowed
-    def test_sdproxy_from_sdapp_to_sdproxy_allowed(self):
-        assert self._policy_exists("sd-app", "sd-proxy", "securedrop.Proxy")
-
-    # securedrop.Proxy from anything else to sd-proxy should be denied
-    def test_sdproxy_from_other_to_sdproxy_denied(self):
-        assert not self._policy_exists("sys-net", "sd-proxy", "securedrop.Proxy")
-        assert not self._policy_exists("sys-firewall", "sd-proxy", "securedrop.Proxy")
-
-    # qubes.Gpg, qubes.GpgImportKey, and qubes.Gpg2 from anything else to sd-gpg should be denied
-    def test_qubesgpg_from_other_to_sdgpg_denied(self):
-        assert not self._policy_exists("sys-net", "sd-gpg", "qubes.Gpg")
-        assert not self._policy_exists("sys-firewall", "sd-gpg", "qubes.Gpg")
-        assert not self._policy_exists("sys-net", "sd-gpg", "qubes.GpgImportKey")
-        assert not self._policy_exists("sys-firewall", "sd-gpg", "qubes.GpgImportKey")
-        assert not self._policy_exists("sys-net", "sd-gpg", "qubes.Gpg2")
-        assert not self._policy_exists("sys-firewall", "sd-gpg", "qubes.Gpg2")
-
-
-def load_tests(loader, tests, pattern):
-    return unittest.TestLoader().loadTestsFromTestCase(SD_Qubes_Rpc_Tests)
+
+
+@functools.cache
+def qrexec_policy_graph(service):
+    cmd = ["qrexec-policy-graph", "--service", service]
+    p = subprocess.run(cmd, capture_output=True, text=True, check=False)
+    return p.stdout
+
+
+def policy_exists(source, target, service):
+    service_policy_graph = qrexec_policy_graph(service)
+    policy_str = f'"{source}" -> "{target}" [label="{service}"'
+    return policy_str in service_policy_graph
+
+
+def test_policy_files_exist():
+    """verify the policies are installed"""
+    assert os.path.exists("/etc/qubes/policy.d/31-securedrop-workstation.policy")
+    assert os.path.exists("/etc/qubes/policy.d/32-securedrop-workstation.policy")
+
+
+# securedrop.Log from @tag:sd-workstation to sd-log should be allowed
+def test_sdlog_from_sdw_to_sdlog_allowed(sdw_tagged_vms):
+    for vm in sdw_tagged_vms:
+        if vm.name != "sd-log":
+            assert policy_exists(vm, "sd-log", "securedrop.Log")
+
+
+# securedrop.Log from anything else to sd-log should be denied
+def test_sdlog_from_other_to_sdlog_denied(all_vms, sdw_tagged_vms):
+    non_sd_workstation_vms = set(all_vms).difference(set(sdw_tagged_vms))
+    for vm in non_sd_workstation_vms:
+        if vm.name != "sd-log":
+            assert not policy_exists(vm, "sd-log", "securedrop.Log")
+
+
+# securedrop.Proxy from sd-app to sd-proxy should be allowed
+def test_sdproxy_from_sdapp_to_sdproxy_allowed():
+    assert policy_exists("sd-app", "sd-proxy", "securedrop.Proxy")
+
+
+# securedrop.Proxy from anything else to sd-proxy should be denied
+def test_sdproxy_from_other_to_sdproxy_denied():
+    assert not policy_exists("sys-net", "sd-proxy", "securedrop.Proxy")
+    assert not policy_exists("sys-firewall", "sd-proxy", "securedrop.Proxy")
+
+
+# qubes.Gpg, qubes.GpgImportKey, and qubes.Gpg2 from anything else to sd-gpg should be denied
+def test_qubesgpg_from_other_to_sdgpg_denied():
+    assert not policy_exists("sys-net", "sd-gpg", "qubes.Gpg")
+    assert not policy_exists("sys-firewall", "sd-gpg", "qubes.Gpg")
+    assert not policy_exists("sys-net", "sd-gpg", "qubes.GpgImportKey")
+    assert not policy_exists("sys-firewall", "sd-gpg", "qubes.GpgImportKey")
+    assert not policy_exists("sys-net", "sd-gpg", "qubes.Gpg2")
+    assert not policy_exists("sys-firewall", "sd-gpg", "qubes.Gpg2")

tests/test_qubes_vms.py

@@ -1,42 +1,31 @@
-import unittest
-
-from qubesadmin import Qubes
-
 from tests.base import CURRENT_FEDORA_DVM, CURRENT_FEDORA_TEMPLATE
 
+"""
+Ensures that the upstream, Qubes-maintained VMs are
+sufficiently up to date.
+"""
 
-class SD_Qubes_VM_Tests(unittest.TestCase):
+
+def test_current_fedora_for_sys_vms(all_vms):
     """
-    Ensures that the upstream, Qubes-maintained VMs are
-    sufficiently up to date.
+    Checks that all sys-* VMs are configured to use
+    an up-to-date version of Fedora.
     """
-
-    def setUp(self):
-        self.app = Qubes()
-
-    def tearDown(self):
-        pass
-
-    def test_current_fedora_for_sys_vms(self):
-        """
-        Checks that all sys-* VMs are configured to use
-        an up-to-date version of Fedora.
-        """
-        sys_vms = ["sys-firewall", "sys-net", "sys-usb", "default-mgmt-dvm"]
-        sys_vms_maybe_disp = ["sys-firewall", "sys-usb"]
-        sys_vms_custom_disp = ["sys-usb"]
-
-        for sys_vm in sys_vms:
-            vm = self.app.domains[sys_vm]
-            wanted_templates = [CURRENT_FEDORA_TEMPLATE]
-            if sys_vm in sys_vms_maybe_disp:
-                if sys_vm in sys_vms_custom_disp:
-                    wanted_templates.append(f"sd-{CURRENT_FEDORA_DVM}")
-                else:
-                    wanted_templates.append(CURRENT_FEDORA_DVM)
-
-            assert vm.template.name in wanted_templates, (
-                f"Unexpected template for {sys_vm}\n"
-                + f"Current: {vm.template.name}\n"
-                + "Expected: {}".format(", ".join(wanted_templates))
-            )
+    sys_vms = ["sys-firewall", "sys-net", "sys-usb", "default-mgmt-dvm"]
+    sys_vms_maybe_disp = ["sys-firewall", "sys-usb"]
+    sys_vms_custom_disp = ["sys-usb"]
+
+    for sys_vm in sys_vms:
+        vm = all_vms[sys_vm]
+        wanted_templates = [CURRENT_FEDORA_TEMPLATE]
+        if sys_vm in sys_vms_maybe_disp:
+            if sys_vm in sys_vms_custom_disp:
+                wanted_templates.append(f"sd-{CURRENT_FEDORA_DVM}")
+            else:
+                wanted_templates.append(CURRENT_FEDORA_DVM)
+
+        assert vm.template.name in wanted_templates, (
+            f"Unexpected template for {sys_vm}\n"
+            + f"Current: {vm.template.name}\n"
+            + "Expected: {}".format(", ".join(wanted_templates))
+        )