Merge remote-tracking branch 'origin/develop' into HEAD

oci/tests/test_catalog_web_links.py

@@ -0,0 +1,102 @@
+"""Catalog links agree with the configured service and preserve suite choices."""
+
+import copy
+import json
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+sys.path.insert(0, str(ROOT / "remote"))
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+from proxmenux_oci.catalog import Catalog
+from proxmenux_oci.installer import DEFAULT_MODE, build_deployment
+from proxmenux_oci.stack import StackError
+from test_advanced_flow_order import RecordingUI, addresses, storages
+import oci_description
+import oci_instance_transaction as transaction
+
+
+class CatalogWebLinks(unittest.TestCase):
+    catalog = Catalog(ROOT)
+
+    def test_jdownloader_launch_matches_its_secure_web_service(self):
+        template = self.catalog.compose("jdownloader")
+        environment = {entry["name"]: entry.get("example")
+                       for entry in template["container_contract"]["environment"]}
+        self.assertEqual(environment["SECURE_CONNECTION"], "1")
+        launch = template["catalog_ui"]["launch"]
+        self.assertEqual(launch, {"scheme": "https", "port": 5800, "path": "/"})
+        endpoint = template["first_run"]["endpoints"][0]
+        self.assertEqual({key: endpoint[key] for key in launch}, launch)
+        self.assertIn('href="https://192.0.2.10:5800/"',
+                      oci_description.render(template, "", "test-installation", "192.0.2.10"))
+
+    def test_shipped_templates_keep_the_curated_link_contract(self):
+        for app in ("jdownloader", "suite-arr"):
+            with self.subTest(app=app):
+                curated = json.loads((ROOT / "catalog/curated" / (app + ".json")).read_text())
+                generated = json.loads((ROOT / "catalog/apps" / (app + ".json")).read_text())
+                self.assertEqual(curated["catalog_ui"]["launch"], generated["catalog_ui"]["launch"])
+                self.assertEqual(curated["first_run"], generated["first_run"])
+
+    def test_suite_parent_names_the_documented_prowlarr_web_endpoint(self):
+        template = self.catalog.compose("suite-arr")
+        endpoints = template["first_run"]["endpoints"]
+        self.assertEqual(len(endpoints), 1)
+        endpoint = endpoints[0]
+        self.assertEqual((endpoint["scheme"], endpoint["port"], endpoint["path"]),
+                         ("http", 9696, "/"))
+        self.assertEqual(endpoint["label"], "Web UI")
+        self.assertNotEqual(endpoint["source"], transaction.GUESSED_ENDPOINT)
+        check = transaction.effective_healthcheck(template)
+        self.assertEqual((check["scheme"], check["port"]), ("http", 9696))
+        self.assertNotIn("accept_non_http", check)
+
+    @patch("proxmenux_oci.installer.host.gpus", return_value={"intel": [], "amd": [], "nvidia": False})
+    @patch("proxmenux_oci.i18n.language", return_value="en")
+    @patch("proxmenux_oci.installer.host.storages", side_effect=storages)
+    @patch("proxmenux_oci.installer.host.bridges", return_value=[{"iface": "vmbr0", "cidr": "192.0.2.1/24"}])
+    @patch("proxmenux_oci.installer.host.timezone", return_value="Etc/UTC")
+    @patch("proxmenux_oci.installer.access.ask_ipv4", return_value=("dhcp", None))
+    @patch("proxmenux_oci.installer.access.ask_addresses", side_effect=addresses)
+    def test_suite_members_keep_their_own_links_with_or_without_prowlarr(self, *_):
+        template = self.catalog.compose("suite-arr")
+        before = copy.deepcopy(template)
+        with patch("subprocess.run") as command:
+            for selected in (["prowlarr", "sonarr"], ["sonarr"]):
+                ui = RecordingUI({"Arr suite: applications to install": selected,
+                                  "Media server": "none"})
+                plan = build_deployment(copy.deepcopy(template), ui, DEFAULT_MODE)
+                self.assertEqual([member["name"] for member in plan["services"]], selected)
+                for member in plan["services"]:
+                    port = 9696 if member["name"] == "prowlarr" else 8989
+                    endpoint = member["template"]["first_run"]["endpoints"][0]
+                    self.assertEqual(endpoint["port"], port)
+                    self.assertEqual(member["healthcheck"]["endpoint"]["port"], port)
+                    notes = oci_description.render(member["template"], "", "test-installation", "192.0.2.10")
+                    self.assertIn(f':{port}/"', notes)
+                    if port != 9696:
+                        self.assertNotIn(':9696/"', notes)
+            command.assert_not_called()
+        self.assertEqual(template, before)
+
+    def test_empty_suite_selection_has_no_host_or_runtime_effect(self):
+        template = self.catalog.compose("suite-arr")
+        before = copy.deepcopy(template)
+        ui = RecordingUI({"Arr suite: applications to install": [], "Media server": "none"})
+        with patch("proxmenux_oci.i18n.language", return_value="en"), \
+                patch("proxmenux_oci.installer.host.storages") as host, \
+                patch("subprocess.run") as command:
+            with self.assertRaises(StackError):
+                build_deployment(template, ui, DEFAULT_MODE)
+            host.assert_not_called()
+            command.assert_not_called()
+        self.assertEqual(template, before)
+
+
+if __name__ == "__main__":
+    unittest.main()

oci/tests/test_managed_volume_installer.py

@@ -0,0 +1,162 @@
+"""Exercise actual installer and persisted hook entrypoints with local fixtures."""
+from pathlib import Path
+import json
+import os
+import shlex
+import subprocess
+import sys
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class FirstInstallTests(unittest.TestCase):
+    def run_case(self, mount_type='managed-volume', read_only=False, stack=False,
+                 invalid=False, empty=False):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp); rootfs = base / 'rootfs'
+            lost = rootfs / 'config/lost+found'; lost.mkdir(parents=True, mode=0o700)
+            sentinel = rootfs / 'config/keep'; sentinel.write_text('application data')
+            deployment = {'stack_managed': stack, 'mounts': [] if empty else [{
+                'type': mount_type, 'container_path': 'relative' if invalid else '/config',
+                'backup': True, 'read_only': read_only}]}
+            plan = base / 'deployment.json'; plan.write_text(json.dumps(deployment))
+            wrapper = base / 'python3'
+            wrapper.write_text('#!' + sys.executable + '\n' +
+                'import sys\nsys.path[:0]=' + repr([str(ROOT / 'remote'), str(ROOT / 'tests')]) + '\n' +
+                'from test_managed_volume_startup import ext4_metadata\n' +
+                'import oci_managed_volumes\nsys.argv=sys.argv[1:]\n' +
+                'with ext4_metadata():\n raise SystemExit(oci_managed_volumes.main())\n')
+            wrapper.chmod(0o700)
+            source = (ROOT / 'remote/install_oci.sh').read_text()
+            start = source.index('prepare_managed_format_directories() {')
+            end = source.index('\nprepare_managed_format_directories "/var/lib/lxc/', start)
+            function = source[start:end]
+            script = '''set -Eeuo pipefail
+VMID=1000
+MOUNT_ENTRIES=$COUNT
+mount_ct_rootfs() { printf 'MOUNT\n'; }
+oci_quiet() {
+  if [[ $1 == pct ]]; then printf 'UNMOUNT\n'; else "$@"; fi
+}
+translate() { printf '%s' "$1"; }
+die() { printf 'FAILURE\n'; exit 70; }
+''' + function + '\nprepare_managed_format_directories "$ROOTFS"\n'
+            result = subprocess.run(['bash', '-c', script], capture_output=True, text=True, timeout=10,
+                env=dict(os.environ, PATH=str(base) + ':' + os.environ['PATH'],
+                         SCRIPT_DIR=str(ROOT / 'remote'), DEPLOYMENT_FILE=str(plan),
+                         ROOTFS=str(rootfs), COUNT='0' if empty else '1'))
+            self.assertEqual(sentinel.read_text(), 'application data')
+            return result, lost.exists()
+
+    def test_unannotated_config_volume_is_prepared_before_first_start(self):
+        result, exists = self.run_case()
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        self.assertFalse(exists)
+        self.assertEqual(result.stdout.splitlines(), ['MOUNT', 'UNMOUNT'])
+
+    def test_readonly_and_bind_sources_are_unchanged(self):
+        for options in ({'read_only': True}, {'mount_type': 'host-bind'}):
+            with self.subTest(options=options):
+                result, exists = self.run_case(**options)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertTrue(exists)
+
+    def test_cleanup_failure_unmounts_and_stops_before_start(self):
+        result, exists = self.run_case(invalid=True)
+        self.assertEqual(result.returncode, 70)
+        self.assertEqual(result.stdout.splitlines(), ['MOUNT', 'UNMOUNT', 'FAILURE'])
+        self.assertTrue(exists)
+
+    def test_stack_member_and_empty_mount_list_have_no_mount_or_cleanup_calls(self):
+        for options in ({'stack': True}, {'empty': True}):
+            with self.subTest(options=options):
+                result, exists = self.run_case(**options)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, '')
+                self.assertTrue(exists)
+
+
+class PersistedHookTests(unittest.TestCase):
+    def run_case(self, *, failure=False, helper=True, invalid=False, timeout=False):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp); marker = base / '1000.console.log'; events = base / 'events'
+            source = (ROOT / 'remote/oci_console_mark.sh').read_text()
+            source = source.replace('/var/log/proxmenux/oci/', str(base) + '/')
+            script = base / 'oci_console_mark.sh'; script.write_text(source)
+            (base / 'oci_dhcp_lease.py').touch()
+            if helper: (base / 'oci_managed_volumes.py').touch()
+            python = base / 'python3'
+            python.write_text('#!/bin/sh\n' +
+                'test ! -f "$MARKER" || exit 80\n' +
+                'printf "cleanup %s\\n" "$3" >> "$EVENTS"\n' +
+                ('sleep 30\n' if timeout else '') + ('exit 1\n' if failure else 'exit 0\n'))
+            python.chmod(0o700)
+            timer = base / 'systemd-run'
+            timer.write_text('#!/bin/sh\ntest -s "$MARKER" || exit 81\nprintf "dhcp\\n" >> "$EVENTS"\n')
+            timer.chmod(0o700)
+            result = subprocess.run(['sh', str(script), 'invalid;id' if invalid else '1000'],
+                capture_output=True, text=True, timeout=15,
+                env=dict(os.environ, PATH=str(base) + ':' + os.environ['PATH'],
+                         MARKER=str(marker), EVENTS=str(events)))
+            return result, marker.exists(), events.read_text().splitlines() if events.exists() else []
+
+    def test_cleanup_precedes_console_marker_and_dhcp_timer(self):
+        result, marker, events = self.run_case()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertTrue(marker)
+        self.assertEqual(events, ['cleanup 1000', 'dhcp'])
+
+    def test_missing_or_failed_helper_keeps_marker_and_dhcp(self):
+        for options, expected in (({'helper': False}, ['dhcp']),
+                                  ({'failure': True}, ['cleanup 1000', 'dhcp'])):
+            with self.subTest(options=options):
+                result, marker, events = self.run_case(**options)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertTrue(marker)
+                self.assertEqual(events, expected)
+
+    def test_hung_cleanup_is_bounded_and_dhcp_still_runs(self):
+        result, marker, events = self.run_case(timeout=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertTrue(marker)
+        self.assertEqual(events, ['cleanup 1000', 'dhcp'])
+
+    def test_invalid_vmid_has_no_cleanup_marker_or_timer(self):
+        result, marker, events = self.run_case(invalid=True)
+        self.assertEqual(result.returncode, 0)
+        self.assertFalse(marker)
+        self.assertEqual(events, [])
+
+
+class InstallerTransportTests(unittest.TestCase):
+    def test_local_and_ssh_staging_include_the_guarded_helper_without_host_calls(self):
+        import io
+        import tarfile
+        from unittest.mock import patch
+        sys.path.insert(0, str(ROOT / 'src'))
+        from proxmenux_oci import installer
+        expected = (ROOT / 'remote/oci_managed_volumes.py').read_bytes()
+        with tempfile.TemporaryDirectory() as tmp:
+            template = Path(tmp) / 'template.json'; template.write_text('{}')
+            for transport in ('local', 'unused-test-host'):
+                for kind in (None, 'generic-multi-lxc-stack'):
+                    with self.subTest(transport=transport, kind=kind):
+                        def inspect(command, payload=None):
+                            if transport == 'local':
+                                folder = Path(command[1]).parent
+                                self.assertEqual((folder / 'oci_managed_volumes.py').read_bytes(), expected)
+                            else:
+                                with tarfile.open(fileobj=io.BytesIO(payload), mode='r:gz') as archive:
+                                    self.assertEqual(archive.extractfile('oci_managed_volumes.py').read(), expected)
+                            return {'checked': True}
+                        with patch.object(installer, '_stream_process', side_effect=inspect), \
+                                patch.object(installer.subprocess, 'run') as host:
+                            self.assertEqual(installer._run_remote_install(ROOT, template,
+                                {'deployment_kind': kind}, transport, dry_run=True), {'checked': True})
+                            host.assert_not_called()
+
+
+if __name__ == '__main__':
+    unittest.main()

oci/tests/test_managed_volume_startup.py

@@ -0,0 +1,326 @@
+"""Managed filesystem preparation must preserve read-only data."""
+from pathlib import Path
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / 'remote'))
+import oci_instance_transaction as transaction
+
+
+class RecoveryGuardTests(unittest.TestCase):
+    def test_read_only_managed_volume_is_not_probed_or_changed(self):
+        deployment = {'mounts': [{'type': 'managed-volume', 'backup': True,
+                                  'container_path': '/config', 'read_only': True}]}
+        with patch.object(Path, 'is_symlink', return_value=False) as probe, \
+                patch.object(Path, 'exists', return_value=False):
+            transaction.remove_empty_format_directories('/fixture/rootfs', deployment)
+        probe.assert_not_called()
+
+    def test_nested_read_only_mount_protects_the_format_directory(self):
+        deployment = {'mounts': [
+            {'type': 'managed-volume', 'backup': True, 'container_path': '/config'},
+            {'type': 'host-bind', 'container_path': '/config/lost+found', 'read_only': True},
+        ]}
+        with patch.object(Path, 'is_symlink', return_value=False) as probe, \
+                patch.object(Path, 'exists', return_value=False):
+            transaction.remove_empty_format_directories('/fixture/rootfs', deployment)
+        probe.assert_not_called()
+
+    def test_a_relative_mount_path_is_rejected_before_filesystem_probes(self):
+        deployment = {'mounts': [{'type': 'managed-volume', 'backup': True,
+                                  'container_path': 'config'}]}
+        with patch.object(Path, 'is_symlink', return_value=False) as probe, \
+                patch.object(Path, 'exists', return_value=False):
+            with self.assertRaises(ValueError):
+                transaction.remove_empty_format_directories('/fixture/rootfs', deployment)
+        probe.assert_not_called()
+
+
+import contextlib
+import copy
+import fcntl
+import json
+import os
+from types import SimpleNamespace
+import subprocess
+
+import oci_managed_volumes as volumes
+import oci_restore_recovery as recovery
+import oci_instances as instances
+
+
+@contextlib.contextmanager
+def ext4_metadata(**overrides):
+    """Use real filesystem operations; only mkfs inode/ownership are fixtures."""
+    real_stat, real_fstat = os.stat, os.fstat
+    def info(value):
+        fields = {key: getattr(value, key) for key in ('st_dev', 'st_ino', 'st_uid', 'st_gid', 'st_mode')}
+        fields.update(st_ino=11, st_uid=0, st_gid=0)
+        fields.update(overrides)
+        return SimpleNamespace(**fields)
+    def observed(path, *args, **kwargs):
+        value = real_stat(path, *args, **kwargs)
+        return info(value) if str(path) == 'lost+found' else value
+    def opened(fd):
+        value = real_fstat(fd)
+        return info(value) if os.readlink(f'/proc/self/fd/{fd}').endswith('/lost+found') else value
+    with patch.object(volumes.os, 'stat', side_effect=observed), \
+            patch.object(volumes.os, 'fstat', side_effect=opened):
+        yield
+
+
+class FormatDirectoryTests(unittest.TestCase):
+    def fixture(self, root):
+        target = root / 'config'
+        target.mkdir()
+        lost = target / 'lost+found'
+        lost.mkdir(mode=0o700)
+        (target / 'keep').write_bytes(b'application data')
+        return lost, {'mounts': [{'container_path': '/config', 'type': 'managed-volume', 'backup': True}]}
+
+    def test_only_an_empty_format_directory_is_removed(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); lost, deployment = self.fixture(root)
+            before = copy.deepcopy(deployment)
+            with ext4_metadata():
+                self.assertEqual(volumes.remove_empty_format_directories(root, deployment), 1)
+            self.assertFalse(lost.exists())
+            self.assertEqual((root / 'config/keep').read_bytes(), b'application data')
+            self.assertEqual(deployment, before)
+
+    def test_running_container_root_reached_through_a_link_is_cleaned(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp); real = base / 'rootfs'; real.mkdir()
+            lost, deployment = self.fixture(real)
+            proc_root = base / 'root'; proc_root.symlink_to(real, target_is_directory=True)
+            with ext4_metadata():
+                self.assertEqual(volumes.remove_empty_format_directories(proc_root, deployment), 1)
+            self.assertFalse(lost.exists())
+            self.assertEqual((real / 'config/keep').read_bytes(), b'application data')
+
+    def test_nonempty_format_directory_has_no_remove_call(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); lost, deployment = self.fixture(root)
+            (lost / 'recovered-file').write_bytes(b'must survive')
+            with ext4_metadata(), patch.object(volumes.os, 'rmdir') as remove:
+                self.assertEqual(volumes.remove_empty_format_directories(root, deployment), 0)
+                remove.assert_not_called()
+            self.assertEqual((lost / 'recovered-file').read_bytes(), b'must survive')
+
+    def test_nonstandard_metadata_is_preserved_without_remove(self):
+        for override in ({'st_ino': 12}, {'st_uid': 100000}, {'st_gid': 100000}, {'st_mode': 0o40755}):
+            with self.subTest(override=override), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); lost, deployment = self.fixture(root)
+                with ext4_metadata(**override), patch.object(volumes.os, 'rmdir') as remove:
+                    self.assertEqual(volumes.remove_empty_format_directories(root, deployment), 0)
+                    remove.assert_not_called()
+                self.assertTrue(lost.is_dir())
+
+    def test_link_and_parent_link_never_reach_external_data(self):
+        for parent in (False, True):
+            with self.subTest(parent=parent), tempfile.TemporaryDirectory() as tmp:
+                base = Path(tmp); root = base / 'root'; root.mkdir()
+                outside = base / 'outside'; outside.mkdir()
+                (outside / 'keep').write_text('external')
+                if parent:
+                    (root / 'config').symlink_to(outside, target_is_directory=True)
+                else:
+                    (root / 'config').mkdir()
+                    (root / 'config/lost+found').symlink_to(outside, target_is_directory=True)
+                deployment = {'mounts': [{'container_path': '/config', 'type': 'managed-volume', 'backup': True}]}
+                with patch.object(volumes.os, 'rmdir') as remove:
+                    if parent:
+                        with self.assertRaises(ValueError):
+                            volumes.remove_empty_format_directories(root, deployment)
+                    else:
+                        self.assertEqual(volumes.remove_empty_format_directories(root, deployment), 0)
+                    remove.assert_not_called()
+                self.assertEqual((outside / 'keep').read_text(), 'external')
+
+    def test_bind_readonly_and_nonbacked_recovery_make_no_filesystem_calls(self):
+        for options in ({'type': 'host-bind'}, {'read_only': True}, {'backup': False}):
+            mount = {'container_path': '/config', 'type': 'managed-volume', 'backup': True, **options}
+            with self.subTest(options=options), patch.object(volumes.os, 'open') as opened, \
+                    patch.object(volumes.os, 'rmdir') as remove:
+                self.assertEqual(volumes.remove_empty_format_directories('/rootfs', {'mounts': [mount]}), 0)
+                opened.assert_not_called(); remove.assert_not_called()
+
+    def test_new_nonbacked_managed_disk_is_prepared_without_changing_backup_policy(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); lost, deployment = self.fixture(root)
+            deployment['mounts'][0]['backup'] = False
+            with ext4_metadata():
+                self.assertEqual(volumes.remove_empty_format_directories(root, deployment, backed_only=False), 1)
+            self.assertFalse(lost.exists())
+            self.assertFalse(deployment['mounts'][0]['backup'])
+
+    def test_read_failure_does_not_mean_empty_and_does_not_remove(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp); lost, deployment = self.fixture(root)
+            with ext4_metadata(), patch.object(volumes.os, 'scandir', side_effect=PermissionError), \
+                    patch.object(volumes.os, 'rmdir') as remove:
+                with self.assertRaises(PermissionError):
+                    volumes.remove_empty_format_directories(root, deployment)
+                remove.assert_not_called()
+            self.assertTrue(lost.is_dir())
+
+    def test_invalid_paths_are_rejected_before_any_filesystem_call(self):
+        for path in ('config', '/', '//', '/config/../outside', '/config/./data', '/bad\npath'):
+            with self.subTest(path=path), patch.object(volumes.os, 'open') as opened:
+                with self.assertRaises(ValueError):
+                    volumes.remove_empty_format_directories('/rootfs', {'mounts': [
+                        {'container_path': path, 'type': 'managed-volume', 'backup': True}]})
+                opened.assert_not_called()
+
+
+    def test_duplicate_paths_are_rejected_before_filesystem_calls(self):
+        mount = {'container_path': '/config', 'type': 'managed-volume', 'backup': True}
+        with patch.object(volumes.os, 'open') as opened:
+            with self.assertRaises(ValueError):
+                volumes.remove_empty_format_directories('/rootfs', {'mounts': [mount, dict(mount)]})
+            opened.assert_not_called()
+
+
+    def test_missing_rootfs_is_an_error_without_remove(self):
+        with tempfile.TemporaryDirectory() as tmp, patch.object(volumes.os, 'rmdir') as remove:
+            with self.assertRaises(FileNotFoundError):
+                volumes.remove_empty_format_directories(Path(tmp) / 'missing', {'mounts': [
+                    {'container_path': '/config', 'type': 'managed-volume', 'backup': True}]})
+            remove.assert_not_called()
+
+    def test_missing_managed_mount_is_an_error_without_remove(self):
+        with tempfile.TemporaryDirectory() as tmp, patch.object(volumes.os, 'rmdir') as remove:
+            with self.assertRaises(FileNotFoundError):
+                volumes.remove_empty_format_directories(Path(tmp), {'mounts': [
+                    {'container_path': '/missing', 'type': 'managed-volume', 'backup': True}]})
+            remove.assert_not_called()
+
+
+class StartupContractTests(unittest.TestCase):
+    def setUp(self):
+        directory = tempfile.TemporaryDirectory(); self.addCleanup(directory.cleanup)
+        self.base = Path(directory.name); self.registry = self.base / 'registry'
+        self.registry.mkdir(); (self.registry / '.lock').touch(mode=0o600)
+        self.record = {'schema_version': 1, 'vmid': 1000,
+                       'installation_id': '11111111-1111-4111-8111-111111111111',
+                       'status': 'installed', 'deployment': {'mounts': [
+                           {'container_path': '/config', 'type': 'managed-volume', 'backup': True}]}}
+        (self.registry / '1000').mkdir()
+        self.record_path = self.registry / '1000/oci-compose.json'
+        self.record_path.write_text(json.dumps(self.record))
+        self.rootfs = self.base / 'guests/1000/rootfs'
+        self.rootfs.mkdir(parents=True)
+        self.lost = self.rootfs / 'config/lost+found'; self.lost.mkdir(parents=True, mode=0o700)
+        (self.rootfs / 'config/keep').write_text('application data')
+        self.config = (f'description: <!-- proxmenux-instance={self.record["installation_id"]} -->\n'
+                       'mp0: local:1000/vm-1000-disk-1.raw,mp=/config,backup=1,size=16G\n').encode()
+
+    def call(self, *, config=None, returncode=0):
+        response = SimpleNamespace(returncode=returncode, stdout=self.config if config is None else config)
+        before = self.record_path.read_bytes()
+        with patch.object(volumes.subprocess, 'run', return_value=response) as command:
+            result = volumes.cleanup_startup(self.registry, 1000, self.base / 'guests')
+        self.assertEqual(self.record_path.read_bytes(), before)
+        return result, command
+
+    def test_current_matching_managed_mount_cleans_before_start_without_guest_exec(self):
+        with ext4_metadata():
+            result, command = self.call()
+        self.assertEqual(result, 1); self.assertFalse(self.lost.exists())
+        command.assert_called_once_with(['pct', 'config', '1000'], capture_output=True, timeout=3)
+        self.assertEqual((self.rootfs / 'config/keep').read_text(), 'application data')
+
+    def test_wrong_identity_failed_command_empty_output_and_changed_mount_do_not_remove(self):
+        cases = [(self.config.replace(b'11111111', b'22222222'), 0), (b'', 0), (self.config, 1),
+                 (self.config.replace(b'local:1000/vm-1000-disk-1.raw', b'/host/data'), 0),
+                 (self.config.replace(b'backup=1', b'backup=0'), 0),
+                 (self.config.replace(b'backup=1', b'backup=1,ro=1'), 0)]
+        for config, code in cases:
+            with self.subTest(config=config, code=code), ext4_metadata(), patch.object(volumes.os, 'rmdir') as remove:
+                result, _ = self.call(config=config, returncode=code)
+                self.assertEqual(result, 0); remove.assert_not_called()
+                self.assertTrue(self.lost.is_dir())
+
+    def test_pending_unsupported_or_hostonly_record_never_calls_pct(self):
+        for change in ({'status': 'updating'}, {'status': 'failed'}, {'pending_transaction': '/private/journal'},
+                       {'deployment': {'mounts': [{'container_path': '/config', 'type': 'host-bind'}]}}):
+            with self.subTest(change=change):
+                self.record_path.write_text(json.dumps({**self.record, **change}))
+                result, command = self.call()
+                self.assertEqual(result, 0); command.assert_not_called()
+                self.assertTrue(self.lost.is_dir())
+
+    def test_busy_lock_is_immediate_and_leaves_files_unchanged(self):
+        with (self.registry / '.lock').open('r') as lock:
+            fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            with patch.object(volumes.subprocess, 'run') as command, patch.object(volumes.os, 'rmdir') as remove:
+                with self.assertRaises(BlockingIOError):
+                    volumes.cleanup_startup(self.registry, 1000, self.base / 'guests')
+                command.assert_not_called(); remove.assert_not_called()
+        self.assertTrue(self.lost.is_dir())
+
+    def test_missing_or_invalid_record_and_timeout_never_remove(self):
+        for value in ('missing', 'invalid', 'timeout'):
+            with self.subTest(value=value):
+                if value == 'missing': self.record_path.unlink()
+                elif value == 'invalid': self.record_path.write_text('{')
+                else: self.record_path.write_text(json.dumps(self.record))
+                with patch.object(volumes.subprocess, 'run', side_effect=subprocess.TimeoutExpired('pct', 3)), \
+                        patch.object(volumes.os, 'rmdir') as remove:
+                    with self.assertRaises((OSError, ValueError, subprocess.TimeoutExpired)):
+                        volumes.cleanup_startup(self.registry, 1000, self.base / 'guests')
+                    remove.assert_not_called()
+                self.assertTrue(self.lost.is_dir())
+
+
+class RecoveryConsumerTests(unittest.TestCase):
+    def test_transaction_recovery_uses_real_cleanup_and_always_unmounts(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            base = Path(tmp); root = base / '1000/rootfs'; root.mkdir(parents=True)
+            lost, deployment = FormatDirectoryTests().fixture(root)
+            with patch.object(transaction, 'owned') as owned, patch.object(transaction, 'run') as command, \
+                    patch.object(transaction, 'Path', return_value=base), ext4_metadata(), \
+                    patch.object(volumes, 'remove_empty_format_directories', wraps=volumes.remove_empty_format_directories) as clean:
+                transaction.cleanup_restored_format_dirs(1000, deployment, 'identity')
+                owned.assert_called_once_with(1000, 'identity')
+                self.assertEqual([c.args for c in command.call_args_list],
+                                 [('pct', 'mount', '1000'), ('pct', 'unmount', '1000')])
+                clean.assert_called_once()
+            self.assertFalse(lost.exists())
+            self.assertEqual((root / 'config/keep').read_bytes(), b'application data')
+
+    def test_transaction_failure_still_unmounts_and_identity_failure_never_mounts(self):
+        deployment = {'mounts': [{'type': 'managed-volume', 'container_path': '/config', 'backup': True}]}
+        with patch.object(transaction, 'owned'), patch.object(transaction, 'run') as command, \
+                patch.object(volumes, 'remove_empty_format_directories', side_effect=OSError('read failed')):
+            with self.assertRaises(OSError):
+                transaction.cleanup_restored_format_dirs(1000, deployment, 'identity')
+            self.assertEqual(command.call_args_list[-1].args, ('pct', 'unmount', '1000'))
+        with patch.object(transaction, 'owned', side_effect=ValueError('identity')), \
+                patch.object(transaction, 'run') as command:
+            with self.assertRaises(ValueError):
+                transaction.cleanup_restored_format_dirs(1000, deployment, 'identity')
+            command.assert_not_called()
+
+    def test_recover_menu_consumer_cleans_managed_and_preserves_ro_or_nested_bind(self):
+        for protected in ('none', 'readonly', 'nested-bind'):
+            with self.subTest(protected=protected), tempfile.TemporaryDirectory() as tmp:
+                root = Path(tmp); lost, _ = FormatDirectoryTests().fixture(root)
+                config = 'mp0: local:1000/vm-1000-disk-1.raw,mp=/config,backup=1'
+                if protected == 'readonly': config += ',ro=1'
+                if protected == 'nested-bind': config += '\nmp1: /host/recovery,mp=/config/lost+found,ro=1'
+                with ext4_metadata(), patch.object(recovery.carried, 'container_root', return_value=contextlib.nullcontext(root)), \
+                        patch.object(volumes.os, 'rmdir', wraps=os.rmdir) as remove:
+                    recovery.clean_format_directories({'vmid': 1000, 'config': config})
+                    if protected == 'none':
+                        self.assertFalse(lost.exists()); remove.assert_called_once()
+                    else:
+                        self.assertTrue(lost.is_dir()); remove.assert_not_called()
+                self.assertEqual((root / 'config/keep').read_bytes(), b'application data')
+
+
+if __name__ == '__main__':
+    unittest.main()

oci/tests/test_readonly_volume_preparation.py

@@ -1,7 +1,7 @@
 """Run the installer's volume preparation loop against filesystem fixtures.
 
 Write doubles reject every mutation of the protected directory, including
-lost+found removal. Writable paths still run real rm/chmod; chown is recorded
+lost+found removal. Writable paths keep ordinary directories and run real chmod; chown is recorded
 without requiring root. These tests never invoke Proxmox or touch a container.
 """
 
@@ -62,6 +62,8 @@ TEMPLATE_FILE="$FIXTURE/template.json"
 DEPLOYMENT_FILE="$FIXTURE/deployment.json"
 HOST_BIND_UID=110001; HOST_BIND_GID=110001
 HOST_ROOT_UID=100000; HOST_ROOT_GID=100000
+SCRIPT_DIR="$REMOTE"
+oci_quiet() { printf "WRITE cleanup %s\n" "${!#}"; "$@"; }
 failed=0; volumes_prepared=0
 oci_log() { printf 'log %s\n' "$*"; }
 write_op() {
@@ -87,7 +89,7 @@ test "$failed" = 0
 '''
             result = subprocess.run(
                 ["bash", "-c", script], capture_output=True, text=True,
-                env=dict(os.environ, FIXTURE=str(fixture),
+                env=dict(os.environ, FIXTURE=str(fixture), REMOTE=str(ROOT / "remote"),
                          PROTECTED=str(protected) if read_only else ""),
                 timeout=15,
             )
@@ -102,7 +104,7 @@ test "$failed" = 0
                 self.assertTrue((protected / "lost+found").is_dir())
             elif "prepared=1" in result.stdout:
                 self.assertEqual(protected.stat().st_mode & 0o777, 0o750)
-                self.assertFalse((protected / "lost+found").exists())
+                self.assertTrue((protected / "lost+found").is_dir())
             return result
 
     def assert_skipped(self, result):
@@ -113,7 +115,7 @@ test "$failed" = 0
 
     def assert_prepared(self, result):
         self.assertEqual(result.returncode, 0, result.stderr + result.stdout)
-        for operation in ("rm", "chown", "chmod"):
+        for operation in ("cleanup", "chown", "chmod"):
             self.assertIn(f"WRITE {operation} ", result.stdout)
         self.assertIn("prepared=1", result.stdout)