Merge pull request #1087 from freedomofpress/logs-flow-downstream

Actually test that logs are flowing to sd-log

tests/base.py

@@ -19,6 +19,7 @@
 CURRENT_FEDORA_TEMPLATE = "fedora-" + CURRENT_FEDORA_VERSION + "-xfce"
 CURRENT_FEDORA_DVM = "fedora-" + CURRENT_FEDORA_VERSION + "-dvm"
 CURRENT_WHONIX_VERSION = "17"
+CURRENT_DEBIAN_VERSION = "bookworm"
 
 
 # Lifted from launcher/sdw_util/Util.py

tests/test_log_vm.py

@@ -1,6 +1,9 @@
+import secrets
+import string
+import subprocess
 import unittest
 
-from base import SD_VM_Local_Test
+from base import CURRENT_DEBIAN_VERSION, SD_VM_Local_Test
 
 
 class SD_Log_Tests(SD_VM_Local_Test):
@@ -29,18 +32,40 @@ def test_redis_service_running(self):
         self.assertTrue(self._service_is_active("redis"))
 
     def test_logs_are_flowing(self):
-        cmd_output = self._run("ls -1 /home/user/QubesIncomingLogs")
-        log_dirs = cmd_output.split("\n")
-        # Confirm AppVMs are sending logs
-        self.assertIn("sd-app", log_dirs)
-        # The following will only have logs if the machine has booted,
-        # which is not guaranteed given that we randomize test order.
-        # self.assertIn("sd-devices", log_dirs)
-        # self.assertIn("sd-proxy", log_dirs)
-        # self.assertIn("sd-viewer", log_dirs)
+        """
+        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 self.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"]
+        # VMs we expect logs will not go to
+        no_log_vms = ["sd-gpg", "sd-log", "sd-whonix"]
+
+        # 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:
+                continue
+            # The sudo call will make it into syslog
+            subprocess.check_call(["qvm-run", vm_name, f"sudo echo {token}"])
+
+        for vm_name in all_vms:
+            if vm_name in skip:
+                continue
+            syslog = f"/home/user/QubesIncomingLogs/{vm_name}/syslog.log"
+            if vm_name in no_log_vms:
+                self.assertFalse(self._fileExists(syslog))
+            else:
+                self.assertIn(token, self._get_file_contents(syslog))
 
     def test_log_dirs_properly_named(self):
-        # Rerunning this command to keep test output readable
         cmd_output = self._run("ls -1 /home/user/QubesIncomingLogs")
         log_dirs = cmd_output.split("\n")
         # Confirm we don't have 'host' entries from Whonix VMs