test(oci): isolate cleanup and recovery fixtures from the host

.github/scripts/tests/test_oci_recovery_inventory_wording.py

@@ -75,6 +75,9 @@ class InventoryMessages(unittest.TestCase):
         self.assertNotIn('Host directories that are kept, with their content:', text)
 
     def test_removal_notice_qualifies_saved_record_source(self):
+        # Extracted operational functions must never inspect the real pmxcfs.
+        host = Path(self.enterContext(tempfile.TemporaryDirectory()))
+        inert_path = lambda value: host / Path(value).name if str(value).startswith('/etc/pve/') else Path(value)
         events = []
         scope = {'members_of': lambda root, vmid: (101, {}, [101]),
                  'instances': SimpleNamespace(read=lambda root, vmid: {'installation_id': 'owned'},
@@ -85,7 +88,7 @@ class InventoryMessages(unittest.TestCase):
                  'remove_owned_host_firewall': lambda primary: True,
                  'run': lambda *args: events.append(('run', args)),
                  'subprocess': SimpleNamespace(run=lambda *args, **kwargs: None),
-                 'Path': Path, 'shutil': SimpleNamespace(rmtree=lambda path: None),
+                 'Path': inert_path, 'shutil': SimpleNamespace(rmtree=lambda path: None),
                  'image_cache': SimpleNamespace(prune=lambda root, lock: []),
                  'work_backup': SimpleNamespace(forget=lambda root, vmid: None),
                  'oci_console': SimpleNamespace(remove_log=lambda vmid: events.append(('log', vmid))),

oci/tests/test_backup_carried_record.py

@@ -33,7 +33,8 @@ class BackupCarriedRecordTests(unittest.TestCase):
             code = self.unmount_code if args[1]=='unmount' else 0
             return SimpleNamespace(returncode=code, stdout=value)
         for name, value in [('_run',run),('_host_directory_missing',lambda _:False),
-                            ('ROOTFS',self.rootfs_base),('mapped_root',lambda _: (os.getuid(),os.getgid()))]:
+                            ('ROOTFS',self.rootfs_base),('STACK_CONTRACT',str(Path(temporary.name)/'stack-{}.json')),
+                            ('mapped_root',lambda _: (os.getuid(),os.getgid()))]:
             context=patch.object(carried,name,value);context.start();self.addCleanup(context.stop)
 
     def carry(self, original=None):

oci/tests/test_carried_config_cache.py

@@ -19,6 +19,9 @@ class CarriedConfigCacheTests(unittest.TestCase):
     def setUp(self):
         self.tmp = tempfile.TemporaryDirectory()
         self.addCleanup(self.tmp.cleanup)
+        contract = patch.object(carried, 'STACK_CONTRACT', str(Path(self.tmp.name) / 'stack-{}.json'))
+        contract.start()
+        self.addCleanup(contract.stop)
         self.root = Path(self.tmp.name) / 'instances'
         self.source = Path(self.tmp.name) / 'lxc'
         self.source.mkdir()

oci/tests/test_cluster_paths.py

@@ -1,5 +1,6 @@
 """A migrated container finds what it needs on any node of a cluster."""
 
+import os
 from pathlib import Path
 import sys
 import tempfile
@@ -50,6 +51,7 @@ class SysctlIncludeTests(unittest.TestCase):
     def config(self, include):
         return f"arch: amd64\nlxc.include: {include}\n".encode()
 
+    @unittest.skipUnless(os.geteuid() == 0, "the sysctl include must belong to root")
     def test_cluster_path_and_the_path_of_earlier_installs_are_accepted(self):
         runtime_settings.check(self.config(self.new), DEPLOYMENT, 101)
         runtime_settings.check(self.config(self.old), DEPLOYMENT, 101)
@@ -58,6 +60,7 @@ class SysctlIncludeTests(unittest.TestCase):
         with self.assertRaises(ValueError):
             runtime_settings.check(self.config(self.root / "elsewhere"), DEPLOYMENT, 101)
 
+    @unittest.skipUnless(os.geteuid() == 0, "the sysctl include must belong to root")
     def test_recovery_accepts_both_paths_only(self):
         state = {"vmid": 101, "record": {"deployment": DEPLOYMENT}}
         runtime_settings.check_recovery(self.config(self.new), state)

oci/tests/test_removal_cleanup.py

@@ -45,6 +45,8 @@ class RemovalCleanupTests(unittest.TestCase):
             patch.object(oci_remove, "MONITOR_APPS", self.apps),
             patch.object(oci_remove, "CLUSTER_RECORDS", self.records),
             patch.object(oci_remove, "HOST_MONITOR_INCLUDES", self.host_monitor),
+            patch.object(oci_remove.public_routes, "DIRECTORY", self.root / "public-routes"),
+            patch.object(runtime_settings, "seccomp_path", lambda vmid: self.cluster / f"{vmid}.proxmenux-seccomp"),
             patch.object(runtime_settings, "include_path", lambda vmid: self.cluster / f"{vmid}.sysctls"),
             patch.object(runtime_settings, "legacy_include_path", lambda vmid: self.legacy / f"{vmid}.proxmenux-sysctls"),
             patch.object(oci_remove.subprocess, "run"),

oci/tests/test_stack_public_routes_lifecycle.py

@@ -118,7 +118,14 @@ class ManagedRouteLifecycleTests(unittest.TestCase):
                 root=Path(folder)/'records';root.mkdir()
                 routes.install(deployment(),4901,WORKER)
                 if case=='changed':routes.include_path(4901).write_text('foreign hook')
-                with patch.object(removal.instances,'guest_exists',return_value=case=='reused'), \
+                with patch.multiple(removal,CLUSTER_NODES=Path(folder)/'nodes',
+                     SNIPPETS=Path(folder)/'snippets',MONITOR_APPS=Path(folder)/'apps',
+                     CLUSTER_RECORDS=Path(folder)/'cluster-records'), \
+                     patch.object(removal.runtime_settings,'include_path',lambda vmid:Path(folder)/f'{vmid}.sysctls'), \
+                     patch.object(removal.runtime_settings,'legacy_include_path',lambda vmid:Path(folder)/f'{vmid}.legacy'), \
+                     patch.object(removal.runtime_settings,'seccomp_path',lambda vmid:Path(folder)/f'{vmid}.seccomp'), \
+                     patch.object(removal.oci_console,'LOG_DIR',Path(folder)/'logs'), \
+                     patch.object(removal.instances,'guest_exists',return_value=case=='reused'), \
                      patch.object(removal.instances,'has_contract',return_value=case=='pending'), \
                      patch.object(removal.instances,'read',return_value={'installation_id':WORKER}), \
                      patch.object(removal.instances,'release_orphan',return_value=False), \