fix(oci): prepare managed volumes across stack installs

oci/tests/test_stack_attach_mounts_volume_preparations.py

@@ -0,0 +1,373 @@
+"""Unit tests for the declarative volume_preparations application inside
+install_generic_stack.py's attach_mounts() (Solution A, approved
+2026-09-29): the stack path now runs volume_preparations itself, after
+attach_mounts() has physically attached the real mount — instead of
+relying on install_oci.sh's apply_installer_profile(), whose
+only_when_mount_type check can never match for a stack service because
+create_service() invokes install_oci.sh per-service with an
+always-empty DEPLOYMENT_FILE.mounts.
+
+Corrected after review found two bugs in the first implementation:
+1. resolve_volume_owner() originally computed ONE value shared by both
+   owner_strategy values, so mapped-root incorrectly picked up
+   PUID/PGID/volume_owner (it must always be 100000:100000 or 0:0,
+   independent of the container's application user). It now takes the
+   owner_strategy as an explicit argument and returns a genuinely
+   different value per strategy, mirroring install_oci.sh's
+   HOST_ROOT_UID/GID (mapped-root, fixed) vs. HOST_BIND_UID/GID
+   (mapped-application-user, PUID/PGID + offset).
+2. apply_volume_preparation() no longer chowns directly — it only
+   resolves and returns the declared owner; attach_mounts() applies it
+   AFTER the seed copy and AFTER its own always-run fallback chown
+   (which preserves image ownership for mounts with no declared
+   preparation), so the declared owner is never silently overwritten.
+
+These tests exercise apply_volume_preparation()/resolve_volume_owner()
+directly against a real filesystem fixture (a temp directory standing
+in for the already-mounted container_path) — they do NOT touch pct,
+LXC, or any real container; attach_mounts() itself (which does call
+pct) is exercised only at the metadata/selection level by the existing
+test_stack_volume_preparations.py tests (build_stack() output) and, for
+the post-attach ordering fix, by a source-order guard test below — not
+re-tested here at the pct/LXC level.
+"""
+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"))
+sys.path.insert(0, str(ROOT / "src"))
+
+import install_generic_stack as igs  # noqa: E402
+
+
+def make_service(*, unprivileged=True, environment=None, volume_owner=None,
+                  volume_preparations=None):
+    return {
+        'deployment': {
+            'security': {'unprivileged': unprivileged},
+            'environment': environment or [],
+        },
+        'template': {
+            'proxmox': {
+                'installer_profile': {
+                    **({'volume_owner': volume_owner} if volume_owner else {}),
+                    'volume_preparations': volume_preparations or [],
+                }
+            }
+        },
+    }
+
+
+class ResolveVolumeOwnerTests(unittest.TestCase):
+    def test_mapped_root_unprivileged_is_100000(self):
+        service = make_service(unprivileged=True)
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-root'), (100000, 100000))
+
+    def test_mapped_root_privileged_is_0(self):
+        service = make_service(unprivileged=False)
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-root'), (0, 0))
+
+    def test_mapped_root_ignores_puid_pgid(self):
+        """The bug this guards against: mapped-root with PUID/PGID=1000
+        must still resolve to the fixed root mapping (100000:100000 for
+        unprivileged), NOT the application-user offset (101000:101000)."""
+        service = make_service(unprivileged=True, environment=[
+            {'name': 'PUID', 'value': '1000'},
+            {'name': 'PGID', 'value': '1000'},
+        ])
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-root'), (100000, 100000))
+
+    def test_mapped_root_ignores_volume_owner_override(self):
+        service = make_service(unprivileged=True, volume_owner={'uid': 5000, 'gid': 6000})
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-root'), (100000, 100000))
+
+    def test_mapped_application_user_default_is_offset_zero(self):
+        service = make_service(unprivileged=True)
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (100000, 100000))
+
+    def test_mapped_application_user_privileged_default_is_root(self):
+        service = make_service(unprivileged=False)
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (0, 0))
+
+    def test_mapped_application_user_with_puid_pgid_unprivileged(self):
+        """The bug this guards against, other direction: mapped-root and
+        mapped-application-user with the SAME PUID/PGID=1000 must give
+        DIFFERENT results — this one gets the application offset."""
+        service = make_service(unprivileged=True, environment=[
+            {'name': 'PUID', 'value': '1000'},
+            {'name': 'PGID', 'value': '1000'},
+        ])
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (101000, 101000))
+
+    def test_mapped_application_user_with_puid_pgid_privileged(self):
+        service = make_service(unprivileged=False, environment=[
+            {'name': 'PUID', 'value': '1000'},
+            {'name': 'PGID', 'value': '1000'},
+        ])
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (1000, 1000))
+
+    def test_mapped_application_user_last_matching_environment_name_wins(self):
+        service = make_service(unprivileged=False, environment=[
+            {'name': 'PUID', 'value': '1000'},
+            {'name': 'UID', 'value': '2000'},
+            {'name': 'PGID', 'value': '1000'},
+            {'name': 'GROUP_ID', 'value': '3000'},
+        ])
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (2000, 3000))
+
+    def test_mapped_application_user_invalid_environment_value_is_safely_zero(self):
+        service = make_service(unprivileged=False, environment=[
+            {'name': 'PUID', 'value': 'not-a-number'},
+            {'name': 'PGID', 'value': '-5'},
+        ])
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (0, 0))
+
+    def test_volume_owner_overrides_only_mapped_application_user(self):
+        """volume_owner must change only the application-user strategy,
+        never mapped-root — even when both are resolved for the same
+        service."""
+        service = make_service(unprivileged=False, environment=[
+            {'name': 'PUID', 'value': '1000'},
+            {'name': 'PGID', 'value': '1000'},
+        ], volume_owner={'uid': 5000, 'gid': 6000})
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user'), (5000, 6000))
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-root'), (0, 0))
+
+    def test_invalid_volume_owner_override_is_safely_zero(self):
+        service = make_service(unprivileged=False, environment=[
+            {'name': 'PUID', 'value': '1000'},
+        ], volume_owner={'uid': 'bogus'})
+        self.assertEqual(igs.resolve_volume_owner(service, 'mapped-application-user')[0], 0)
+
+    def test_unknown_owner_strategy_fails_clearly(self):
+        service = make_service(unprivileged=True)
+        with self.assertRaises(RuntimeError):
+            igs.resolve_volume_owner(service, 'some-future-strategy')
+
+
+class ApplyVolumePreparationTests(unittest.TestCase):
+    def _target(self, tmp, with_lost_found=True):
+        target = Path(tmp) / 'target'
+        target.mkdir()
+        if with_lost_found:
+            (target / 'lost+found').mkdir()
+        return target
+
+    def test_managed_volume_preparation_removes_lost_found_and_resolves_owner(self):
+        """apply_volume_preparation() no longer chowns directly — it
+        resolves and returns the declared owner for the caller to apply
+        after the seed copy (see ApplyVolumePreparationOrderingTests)."""
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/app/.blinko',
+                'remove_lost_found': True,
+                'owner_strategy': 'mapped-root',
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/app/.blinko', 'type': 'managed-volume'}
+            applied, owner = igs.apply_volume_preparation(service, mount, target)
+            self.assertTrue(applied)
+            self.assertEqual(owner, (100000, 100000))
+            self.assertFalse((target / 'lost+found').exists())
+
+    def test_host_bind_mount_with_managed_volume_only_rule_is_not_applied(self):
+        """A declared preparation scoped to only_when_mount_type:
+        managed-volume must not run against a host-bind mount, even if
+        the container_path matches."""
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            lost_found_existed_before = (target / 'lost+found').exists()
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/shared/data',
+                'remove_lost_found': True,
+                'owner_strategy': 'mapped-root',
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/shared/data', 'type': 'host-bind'}
+            applied, owner = igs.apply_volume_preparation(service, mount, target)
+            self.assertFalse(applied)
+            self.assertIsNone(owner)
+            self.assertEqual((target / 'lost+found').exists(), lost_found_existed_before)
+
+    def test_final_normalized_postgres_path_is_applied(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/var/lib/postgresql',
+                'remove_lost_found': True,
+                'owner_strategy': 'mapped-root',
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/var/lib/postgresql', 'type': 'managed-volume'}
+            applied, owner = igs.apply_volume_preparation(service, mount, target)
+            self.assertTrue(applied)
+            self.assertEqual(owner, (100000, 100000))
+            self.assertFalse((target / 'lost+found').exists())
+
+    def test_pre_rename_path_does_not_match_final_mount(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/var/lib/postgresql/data',
+                'remove_lost_found': True,
+                'owner_strategy': 'mapped-root',
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/var/lib/postgresql', 'type': 'managed-volume'}
+            applied, owner = igs.apply_volume_preparation(service, mount, target)
+            self.assertFalse(applied)
+            self.assertIsNone(owner)
+            self.assertTrue((target / 'lost+found').exists())
+
+    def test_no_declared_preparation_leaves_caller_to_fall_back(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[])
+            mount = {'container_path': '/app/.blinko', 'type': 'managed-volume'}
+            applied, owner = igs.apply_volume_preparation(service, mount, target)
+            self.assertFalse(applied)
+            self.assertIsNone(owner)
+            self.assertTrue((target / 'lost+found').exists())
+
+    def test_unknown_owner_strategy_fails_clearly(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/app/.blinko',
+                'remove_lost_found': True,
+                'owner_strategy': 'some-future-strategy',
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/app/.blinko', 'type': 'managed-volume'}
+            with self.assertRaises(RuntimeError):
+                igs.apply_volume_preparation(service, mount, target)
+
+    def test_missing_owner_strategy_fails_clearly(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            target = self._target(tmp)
+            service = make_service(unprivileged=True, volume_preparations=[{
+                'container_path': '/app/.blinko',
+                'remove_lost_found': True,
+                'only_when_mount_type': 'managed-volume',
+            }])
+            mount = {'container_path': '/app/.blinko', 'type': 'managed-volume'}
+            with self.assertRaises(RuntimeError):
+                igs.apply_volume_preparation(service, mount, target)
+
+    def test_mapped_application_user_is_resolved_distinctly_from_mapped_root(self):
+        """Integration-level guard for bug #1: the two owner_strategy
+        values applied to the SAME service (same PUID/PGID) must return
+        different owners through apply_volume_preparation() itself, not
+        just through resolve_volume_owner() in isolation."""
+        with tempfile.TemporaryDirectory() as tmp:
+            service = make_service(unprivileged=True, environment=[
+                {'name': 'PUID', 'value': '1000'},
+                {'name': 'PGID', 'value': '1000'},
+            ], volume_preparations=[
+                {
+                    'container_path': '/data/root-owned',
+                    'owner_strategy': 'mapped-root',
+                    'only_when_mount_type': 'managed-volume',
+                },
+                {
+                    'container_path': '/data/app-owned',
+                    'owner_strategy': 'mapped-application-user',
+                    'only_when_mount_type': 'managed-volume',
+                },
+            ])
+            target_root = self._target(tmp, with_lost_found=False)
+            target_app = Path(tmp) / 'target-app'
+            target_app.mkdir()
+            _, owner_root = igs.apply_volume_preparation(
+                service, {'container_path': '/data/root-owned', 'type': 'managed-volume'}, target_root)
+            _, owner_app = igs.apply_volume_preparation(
+                service, {'container_path': '/data/app-owned', 'type': 'managed-volume'}, target_app)
+            self.assertEqual(owner_root, (100000, 100000))
+            self.assertEqual(owner_app, (101000, 101000))
+            self.assertNotEqual(owner_root, owner_app)
+
+
+class AttachMountsOrderingTests(unittest.TestCase):
+    """Guards bug #2 at the source level: attach_mounts() must apply a
+    declared owner_strategy's chown AFTER the seed cp -a and AFTER its
+    own fallback os.chown(target, *owner[:2]) — otherwise that always-run
+    fallback (which exists to preserve image ownership for mounts with
+    no declared preparation) silently overwrites the declared result.
+    This does not execute attach_mounts() (it calls pct/LXC); it proves
+    the source order directly, which is what determines runtime
+    behaviour here."""
+
+    def setUp(self):
+        self.source = (ROOT / "remote" / "install_generic_stack.py").read_text(encoding="utf-8")
+        start = self.source.index("def attach_mounts(")
+        end = self.source.index("\ndef ", start + 1)
+        self.body = self.source[start:end]
+
+    def test_apply_volume_preparation_called_before_seed_copy(self):
+        self.assertLess(
+            self.body.index("apply_volume_preparation("),
+            self.body.index("run('cp','-a',str(seed)"),
+        )
+
+    def test_fallback_chown_runs_before_declared_owner_chown(self):
+        fallback_index = self.body.index("os.chown(target,*owner[:2])")
+        declared_index = self.body.index("os.chown(target, *declared_owner)")
+        self.assertLess(fallback_index, declared_index,
+                         "the declared owner_strategy chown must run AFTER the fallback chown, "
+                         "so it is the mount's final ownership and not silently overwritten")
+
+    def test_declared_owner_chown_is_conditional_on_applied(self):
+        # Must only run when a preparation actually matched — guards
+        # against unconditionally chowning to a stale/None owner.
+        declared_index = self.body.index("os.chown(target, *declared_owner)")
+        guard_index = self.body.rindex("if applied:", 0, declared_index)
+        self.assertGreater(declared_index, guard_index)
+
+
+class InstallOciShStackManagedTests(unittest.TestCase):
+    """install_oci.sh's own volume_preparations loop must be skipped
+    only for stack_managed: true services, and must leave generated_files,
+    self_signed_tls, and every other apply_installer_profile() step
+    untouched — this is a source-text guard (no bash execution here; a
+    real stack install run is exercised separately, outside this unit
+    test file)."""
+
+    def setUp(self):
+        self.source = (ROOT / "remote" / "install_oci.sh").read_text(encoding="utf-8")
+
+    def test_volume_preparations_loop_is_gated_on_stack_managed(self):
+        self.assertIn(
+            'if (( failed == 0 )) && [[ $(jq -r \'.stack_managed // false\' "$DEPLOYMENT_FILE") != true ]]; then',
+            self.source,
+        )
+
+    def test_generated_files_loop_is_not_gated_on_stack_managed(self):
+        marker = "done < <(jq -r '.proxmox.installer_profile.generated_files[]? | @base64' \"$TEMPLATE_FILE\")"
+        self.assertIn(marker, self.source)
+        gate_index = self.source.index(
+            'if (( failed == 0 )) && [[ $(jq -r \'.stack_managed // false\' "$DEPLOYMENT_FILE") != true ]]; then'
+        )
+        self.assertLess(self.source.index(marker), gate_index,
+                         "generated_files loop must run before, and outside, the new stack_managed gate")
+
+    def test_self_signed_tls_block_is_not_gated_on_stack_managed(self):
+        marker = "tls_count == 1"
+        self.assertIn(marker, self.source)
+        gate_index = self.source.index(
+            'if (( failed == 0 )) && [[ $(jq -r \'.stack_managed // false\' "$DEPLOYMENT_FILE") != true ]]; then'
+        )
+        self.assertGreater(self.source.index(marker), gate_index,
+                            "self_signed_tls must remain its own block after the volume_preparations gate, unaffected by it")
+
+    def test_pre_start_repairs_and_volume_seeds_are_not_gated(self):
+        self.assertIn("pre_start_repairs", self.source)
+        self.assertIn("apply_volume_seeds", self.source)
+
+
+if __name__ == "__main__":
+    unittest.main()

oci/tests/test_stack_volume_preparations.py

@@ -0,0 +1,140 @@
+"""Regression test for volume_preparations survival through the generic
+multi-LXC stack pipeline (apply_stack_support + service_template).
+
+install_generic_stack.py::create_service() writes each service's own
+service['template'] to a per-service template.json and runs install_oci.sh
+against that file independently — the parent stack template's
+installer_profile is never read at install time. So the fix must land in
+each service's own template, not just the parent's top-level
+installer_profile (that alone would be a cosmetic data match with no real
+effect). This is the regression guard for the stack volume_preparations
+loss first found via read-only audit (2026-09-29): apply_stack_support()
+replaced installer_profile wholesale and silently dropped
+volume_preparations for every stack app (Blinko, Kimai, Linkwarden, Monica
+Official, Petio, Qui, RomM, Teable).
+"""
+from pathlib import Path
+import sys
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from proxmenux_oci.catalog import Catalog
+from proxmenux_oci.stack import build_stack, DefaultsUI
+
+
+class StackVolumePreparationsTests(unittest.TestCase):
+    def test_stack_service_receives_only_its_own_volume_preparation(self):
+        """Blinko (2-service stack: blinko + blinko-postgres). The real
+        build_stack() output — what create_service() actually writes to
+        each service's template.json — must carry the preparation for
+        /app/.blinko into the blinko service's own template, scoped to
+        only_when_mount_type: managed-volume, and must NOT leak it (or any
+        other service's paths) into blinko-postgres's template, which has
+        no volume_preparations entry of its own in the source catalog."""
+        template = Catalog(ROOT).compose("blinko")
+        result = build_stack(template, DefaultsUI(), mode="simple")
+
+        services_by_name = {s["name"]: s for s in result["services"]}
+        self.assertEqual(set(services_by_name), {"blinko", "blinko-postgres"})
+
+        blinko_preps = services_by_name["blinko"]["template"]["proxmox"] \
+            .get("installer_profile", {}).get("volume_preparations", [])
+        self.assertEqual(len(blinko_preps), 1)
+        self.assertEqual(blinko_preps[0]["container_path"], "/app/.blinko")
+        self.assertEqual(blinko_preps[0]["only_when_mount_type"], "managed-volume")
+
+        postgres_preps = services_by_name["blinko-postgres"]["template"]["proxmox"] \
+            .get("installer_profile", {}).get("volume_preparations", [])
+        self.assertEqual(postgres_preps, [])
+
+    def test_synthetic_dependency_volume_preparation_reaches_only_that_service(self):
+        """Blinko's postgres dependency service (blinko-postgres, image
+        postgres:latest) has its own Compose-declared mount,
+        /var/lib/postgresql/data, which build_stack() normalizes to
+        /var/lib/postgresql for PostgreSQL 18+ (see the 'Official
+        PostgreSQL 18+' comment in build_stack()) — this mount is NOT one
+        of the DEPENDENCY_VOLUMES fallback entries, it comes straight from
+        the source Compose file. A synthetic volume_preparations entry for
+        the final, normalized mount target (/var/lib/postgresql) is
+        injected on the parent template to verify it reaches only the
+        postgres service's own template — not blinko's, and not lost. This
+        test is only about correct metadata transfer through
+        service_template()/build_stack(); it makes no claim about whether
+        preparing a database data directory this way is runtime-appropriate."""
+        template = Catalog(ROOT).compose("blinko")
+        template["proxmox"]["installer_profile"].setdefault("volume_preparations", [])
+        template["proxmox"]["installer_profile"]["volume_preparations"].append({
+            "container_path": "/var/lib/postgresql",
+            "remove_lost_found": True,
+            "owner_strategy": "mapped-root",
+            "only_when_mount_type": "managed-volume",
+        })
+
+        result = build_stack(template, DefaultsUI(), mode="simple")
+        services_by_name = {s["name"]: s for s in result["services"]}
+
+        postgres_preps = services_by_name["blinko-postgres"]["template"]["proxmox"] \
+            .get("installer_profile", {}).get("volume_preparations", [])
+        self.assertEqual(
+            [p["container_path"] for p in postgres_preps],
+            ["/var/lib/postgresql"],
+            "The synthetic preparation for the final, normalized postgres "
+            "mount target must reach the postgres service's own template.",
+        )
+        self.assertEqual(
+            postgres_preps[0]["container_path"],
+            services_by_name["blinko-postgres"]["template"]["container_contract"]["volumes"][0]["container_path"],
+            "The preparation's container_path must match the service's own "
+            "final mount path exactly, not an intermediate/pre-rename form.",
+        )
+
+        blinko_preps = services_by_name["blinko"]["template"]["proxmox"] \
+            .get("installer_profile", {}).get("volume_preparations", [])
+        self.assertEqual(
+            [p["container_path"] for p in blinko_preps],
+            ["/app/.blinko"],
+            "The synthetic postgres-only preparation must not leak into "
+            "blinko's own template.",
+        )
+
+    def test_synthetic_dependency_volume_preparation_keyed_to_the_pre_rename_path_is_also_normalized(self):
+        """The same postgres mount, but the synthetic preparation is keyed
+        to the pre-rename Compose path (/var/lib/postgresql/data) instead
+        of the final one. It must still reach blinko-postgres's own
+        template, remapped to the same final container_path
+        (/var/lib/postgresql) that build_stack() actually mounts — a
+        preparation entry whose path no longer matches any real mount is
+        dead configuration at install time. This test is only about
+        correct metadata transfer; it makes no claim about runtime
+        appropriateness."""
+        template = Catalog(ROOT).compose("blinko")
+        template["proxmox"]["installer_profile"].setdefault("volume_preparations", [])
+        template["proxmox"]["installer_profile"]["volume_preparations"].append({
+            "container_path": "/var/lib/postgresql/data",
+            "remove_lost_found": True,
+            "owner_strategy": "mapped-root",
+            "only_when_mount_type": "managed-volume",
+        })
+
+        result = build_stack(template, DefaultsUI(), mode="simple")
+        services_by_name = {s["name"]: s for s in result["services"]}
+
+        postgres_preps = services_by_name["blinko-postgres"]["template"]["proxmox"] \
+            .get("installer_profile", {}).get("volume_preparations", [])
+        self.assertEqual(
+            [p["container_path"] for p in postgres_preps],
+            ["/var/lib/postgresql"],
+            "A preparation keyed to the pre-rename Compose path must be "
+            "remapped to the same final path build_stack() actually mounts, "
+            "not silently dropped and not left pointing at the stale path.",
+        )
+        self.assertEqual(
+            postgres_preps[0]["container_path"],
+            services_by_name["blinko-postgres"]["template"]["container_contract"]["volumes"][0]["container_path"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

oci/tests/test_volume_preparations_curated.py

@@ -0,0 +1,80 @@
+"""Regression tests for volume_preparations coverage across generated and
+curated-profile catalog entries.
+
+Covers the original OCI lost+found fix: a fresh ext4 managed volume carries
+a lost+found directory, which stops images that take ownership of their own
+data directories. volume_preparations (remove_lost_found +
+only_when_mount_type: managed-volume) must be present for every persistent
+mount, for both regular generated apps/*.json entries and curated-profile
+entries (curated/*.json) — the curated-profile gap (Jellyfin Official first
+surfaced it) is exactly what this suite guards against regressing.
+"""
+from pathlib import Path
+import sys
+import unittest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+from proxmenux_oci.catalog import Catalog
+
+
+class VolumePreparationsCoverageTests(unittest.TestCase):
+    def _preparations_by_path(self, template):
+        preparations = template["proxmox"]["installer_profile"].get("volume_preparations", [])
+        return {item["container_path"]: item for item in preparations}
+
+    def test_generated_app_volume_preparations_match_persistent_mounts(self):
+        """Regular generated app (Uptime Kuma, apps/uptimekuma.json): every
+        declared volume has a matching volume_preparations entry, and every
+        entry is scoped to only_when_mount_type: managed-volume."""
+        template = Catalog(ROOT).compose("uptimekuma")
+
+        volume_paths = {v["container_path"] for v in template["container_contract"].get("volumes", [])}
+        preparations = self._preparations_by_path(template)
+
+        self.assertTrue(volume_paths, "fixture app must declare at least one volume")
+        self.assertEqual(set(preparations.keys()), volume_paths)
+        for path, entry in preparations.items():
+            self.assertTrue(entry.get("remove_lost_found"))
+            self.assertEqual(entry.get("only_when_mount_type"), "managed-volume")
+
+    def test_curated_profile_app_volume_preparations_match_persistent_mounts(self):
+        """Curated-profile app (Jellyfin Official, curated/jellyfin-official.json,
+        3 mounts: /config, /cache, /media): same coverage requirement as
+        generated apps. This is the regression guard for the gap first found
+        via live testing — the original fix only touched apps/*.json and
+        missed curated/*.json entirely."""
+        template = Catalog(ROOT).compose("jellyfin-official")
+
+        volume_paths = {v["container_path"] for v in template["container_contract"].get("volumes", [])}
+        preparations = self._preparations_by_path(template)
+
+        self.assertEqual(volume_paths, {"/config", "/cache", "/media"})
+        self.assertEqual(set(preparations.keys()), volume_paths)
+        for path, entry in preparations.items():
+            self.assertTrue(entry.get("remove_lost_found"))
+            self.assertEqual(entry.get("only_when_mount_type"), "managed-volume")
+
+
+    def test_overlay_app_volume_preparations_match_the_overlay_remapped_mount(self):
+        """PocketBase (apps/pocketbase.json + overlays/pocketbase.json): the
+        overlay remaps the container_path from /pb_data to
+        /pocketbase/pb_data. volume_preparations must follow the overlay's
+        remapped path, not the pre-overlay source path — this is the
+        regression guard for the overlay/mount desync first found via
+        read-only audit (2026-09-29)."""
+        template = Catalog(ROOT).compose("pocketbase")
+
+        volume_paths = {v["container_path"] for v in template["container_contract"].get("volumes", [])}
+        preparations = self._preparations_by_path(template)
+
+        self.assertEqual(volume_paths, {"/pocketbase/pb_data"})
+        self.assertEqual(set(preparations.keys()), volume_paths)
+        for path, entry in preparations.items():
+            self.assertTrue(entry.get("remove_lost_found"))
+            self.assertEqual(entry.get("only_when_mount_type"), "managed-volume")
+
+
+if __name__ == "__main__":
+    unittest.main()