US08-03: Compose the Runtime and Mount the Library Safely (#98)

tests/e2e/test_compose_stack.py

@@ -0,0 +1,337 @@
+"""US08-03: the composition, actually composed.
+
+Nothing here is faked below the process boundary: Docker builds the image, Compose
+starts the migrate/API/worker containers against a temporary fixture library on a
+real bind mount, and every assertion is made over HTTP or against what the stack
+left in its data volume. The vision provider is the deterministic fake seam the
+other end-to-end suites use, because an upload of real photos to a real model is not
+what this story is about — the mount, the lock, the volume, and the restart are.
+
+The file contract (one API, one worker, migrations first, no committed values) is
+checked without a daemon in ``tests/integration/test_compose_runtime.py``; only the
+running proof needs Docker, and CI is where it runs unskipped (US08-04).
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import time
+from pathlib import Path
+
+import httpx
+import pytest
+from PIL import Image
+
+REPO = Path(__file__).resolve().parents[2]
+PROJECT = "photo-pipeline-us0803"
+IMAGE = "photo-pipeline-test:us08-03"
+SECRET = "compose-acceptance-secret"
+CONTAINER_LIBRARY = "/library"
+READY_TIMEOUT_SECONDS = 180
+JOB_TIMEOUT_SECONDS = 180
+UP_TIMEOUT_SECONDS = 30 * 60
+
+pytestmark = pytest.mark.container
+
+
+def compose_available() -> bool:
+    try:
+        return (
+            subprocess.run(
+                ["docker", "compose", "version"], capture_output=True, timeout=60
+            ).returncode
+            == 0
+        )
+    except (OSError, subprocess.SubprocessError):
+        return False
+
+
+needs_compose = pytest.mark.skipif(
+    not compose_available(), reason="no Docker daemon with the compose plugin"
+)
+
+
+def free_port() -> int:
+    with socket.socket() as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+class Stack:
+    """The composition under test, plus the environment it was started with."""
+
+    def __init__(self, library: Path, env_file: Path) -> None:
+        self.library = library
+        self.port = free_port()
+        self.base = f"http://127.0.0.1:{self.port}"
+        # Compose reads the repository's own .env for substitution; the process
+        # environment wins over it, so the test's values are the ones that apply.
+        self.env = {
+            **os.environ,
+            "PHOTO_PIPELINE_IMAGE": IMAGE,
+            "PHOTO_PIPELINE_ENV_FILE": str(env_file),
+            "PHOTO_PIPELINE_LIBRARY_HOST_PATH": str(library),
+            "PHOTO_PIPELINE_LIBRARY_ROOTS": CONTAINER_LIBRARY,
+            "PHOTO_PIPELINE_PORT": str(self.port),
+            "PHOTO_PIPELINE_UID": str(os.getuid()),
+            "PHOTO_PIPELINE_GID": str(os.getgid()),
+        }
+
+    def compose(self, *args: str, check: bool = True, timeout: int = 300):
+        result = subprocess.run(
+            ["docker", "compose", "-p", PROJECT, "-f", str(REPO / "docker-compose.yml"), *args],
+            capture_output=True,
+            text=True,
+            env=self.env,
+            cwd=REPO,
+            timeout=timeout,
+        )
+        if check and result.returncode != 0:
+            raise AssertionError(
+                f"docker compose {' '.join(args)} failed:\n{result.stdout}\n{result.stderr}\n"
+                f"{self.compose('logs', '--tail', '80', check=False).stdout}"
+            )
+        return result
+
+    def wait_until_ready(self) -> None:
+        deadline = time.monotonic() + READY_TIMEOUT_SECONDS
+        while time.monotonic() < deadline:
+            try:
+                if httpx.get(f"{self.base}/api/v1/health/ready", timeout=5).status_code == 200:
+                    return
+            except httpx.HTTPError:
+                pass
+            time.sleep(0.5)
+        logs = self.compose("logs", "--tail", "120", check=False)
+        raise AssertionError(f"the stack never became ready:\n{logs.stdout}\n{logs.stderr}")
+
+    def client(self) -> httpx.Client:
+        client = httpx.Client(base_url=f"{self.base}/api/v1", timeout=60)
+        bootstrap = client.get("/session", headers={"X-Access-Secret": SECRET})
+        assert bootstrap.status_code == 200, bootstrap.text
+        client.headers["X-CSRF-Token"] = bootstrap.json()["csrf_token"]
+        return client
+
+
+@pytest.fixture(scope="module")
+def library() -> Path:
+    """A small fixture library on the host, mounted into both containers.
+
+    Not under pytest's ``tmp_path``: on macOS that is ``/var/folders/...``, which a
+    Docker VM (Colima, Docker Desktop) does not share, so the bind mount would arrive
+    empty and every assertion below would be about nothing. ``$HOME`` is shared by
+    every default configuration.
+    """
+    base = Path(
+        os.environ.get("PHOTO_PIPELINE_TEST_MOUNT_BASE", Path.home() / ".cache" / "photo-pipeline")
+    )
+    base.mkdir(parents=True, exist_ok=True)
+    root = Path(tempfile.mkdtemp(prefix="library-", dir=base))
+    for album, count in (("01_day", 2), ("02_night", 1)):
+        (root / album).mkdir()
+        for index in range(count):
+            colour = (40 * (index + 1), 90, 160)
+            Image.new("RGB", (64, 48), colour).save(root / album / f"{album}_{index}.jpg")
+    # The exclusion sentinel: it must never be discovered, counted, or analyzed.
+    (root / "_IGNORE").mkdir()
+    Image.new("RGB", (32, 32), (0, 0, 0)).save(root / "_IGNORE" / "sentinel.jpg")
+    try:
+        yield root
+    finally:
+        shutil.rmtree(root, ignore_errors=True)
+
+
+@pytest.fixture(scope="module")
+def env_file(tmp_path_factory) -> Path:
+    """Configuration and secrets come from the environment, so the test writes its
+    own file rather than borrowing the operator's."""
+    path = tmp_path_factory.mktemp("config") / "compose.env"
+    path.write_text(
+        "\n".join(
+            [
+                f"PHOTO_PIPELINE_ACCESS_SECRET={SECRET}",
+                "PHOTO_PIPELINE_LOG_FORMAT=text",
+                # The deterministic vision seam, in the data volume so both roles and
+                # the test can see it (concept §18).
+                "PHOTO_PIPELINE_FAKE_VISION_LOG=/data/vision.log",
+            ]
+        )
+        + "\n"
+    )
+    return path
+
+
+@pytest.fixture(scope="module")
+def stack(library, env_file):
+    if not compose_available():
+        pytest.skip("no Docker daemon with the compose plugin")
+    running = Stack(library, env_file)
+    running.compose("down", "--volumes", "--remove-orphans", check=False)
+    running.compose("up", "--detach", "--build", timeout=UP_TIMEOUT_SECONDS)
+    try:
+        running.wait_until_ready()
+        yield running
+    finally:
+        running.compose("down", "--volumes", "--remove-orphans", check=False, timeout=300)
+
+
+def await_job(client: httpx.Client, job_id: str, states=("succeeded",)) -> dict:
+    deadline = time.monotonic() + JOB_TIMEOUT_SECONDS
+    snapshot: dict = {}
+    while time.monotonic() < deadline:
+        response = client.get(f"/jobs/{job_id}")
+        if response.status_code == 200:
+            snapshot = response.json()
+            if snapshot["state"] in states:
+                return snapshot
+        time.sleep(0.5)
+    raise AssertionError(f"job {job_id} never reached {states}: {snapshot}")
+
+
+# ── the mounted library ──────────────────────────────────────────────────────
+
+
+@needs_compose
+def test_the_stack_scans_the_bind_mounted_library_at_its_container_paths(stack):
+    client = stack.client()
+    try:
+        scanned = client.post("/inventory/scan")
+        assert scanned.status_code == 200, scanned.text
+        assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
+    finally:
+        client.close()
+
+    assert len(assets) == 3, assets
+    paths = {asset["current_path"] for asset in assets}
+    assert all(path.startswith(CONTAINER_LIBRARY + "/") for path in paths), paths
+    assert not any("_IGNORE" in path or "sentinel" in path for path in paths)
+    # The host paths are what the operator mounted, and they are not what the
+    # application records: the roots are the container's.
+    assert not any(str(stack.library) in path for path in paths)
+
+
+@needs_compose
+def test_a_library_root_that_is_not_mounted_is_refused_at_startup(stack):
+    """The container-specific failure: configured roots that name nothing mounted."""
+    refused = stack.compose(
+        "run",
+        "--rm",
+        "--no-deps",
+        "--env",
+        "PHOTO_PIPELINE_LIBRARY_ROOTS=/srv/photos",
+        "api",
+        "serve",
+        check=False,
+    )
+    assert refused.returncode == 5, refused.stdout + refused.stderr
+    assert "library root /srv/photos" in refused.stdout + refused.stderr
+
+
+@needs_compose
+def test_a_second_worker_is_refused_by_the_library_lock(stack):
+    """Not by convention: the running worker's lock is in the shared data volume."""
+    refused = stack.compose(
+        "run", "--rm", "--no-deps", "worker", "worker", "--id", "worker-2", check=False
+    )
+    assert refused.returncode == 2, refused.stdout + refused.stderr
+    assert "worker is already running" in refused.stdout + refused.stderr
+
+
+# ── restart, resume, and the data volume ─────────────────────────────────────
+
+
+@needs_compose
+def test_a_queued_job_resumes_after_both_containers_restart(stack):
+    client = stack.client()
+    try:
+        client.post("/inventory/scan").raise_for_status()
+        assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
+        for asset in assets:
+            decided = client.post(
+                "/safety/decisions", json={"asset_id": asset["id"], "decision": "sfw"}
+            )
+            assert decided.status_code == 200, decided.text
+
+        # Stop the worker first, so the job is provably still queued when the restart
+        # happens: a job that finished before the restart would prove nothing.
+        stack.compose("stop", "worker")
+        job = client.post("/analysis/jobs").json()
+        assert client.get(f"/jobs/{job['id']}").json()["state"] == "queued"
+    finally:
+        client.close()
+
+    stack.compose("restart", "api", "worker")
+    stack.wait_until_ready()
+
+    client = stack.client()  # the session is per API process, so it is re-bootstrapped
+    try:
+        finished = await_job(client, job["id"])
+        assert finished["state"] == "succeeded", finished
+        # The database is intact and the work is durable, not merely reported.
+        after = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
+        assert {asset["id"] for asset in after} == {asset["id"] for asset in assets}
+        analysed = client.get(f"/analysis/results/{assets[0]['id']}")
+        assert analysed.status_code == 200, analysed.text
+        assert analysed.json()["description"]
+        # Only the mounted library's own photos were analysed: the sentinel under
+        # _IGNORE is not an asset, so it can never have become an item of this job.
+        assert finished["progress"]["total"] == len(assets)
+    finally:
+        client.close()
+
+
+@needs_compose
+def test_the_data_volume_survives_recreating_the_containers(stack):
+    """`down` without `--volumes` then `up` is the upgrade path: state stays."""
+    client = stack.client()
+    try:
+        client.post("/inventory/scan").raise_for_status()
+        before = {a["id"] for a in client.get("/inventory/assets").json()["items"]}
+    finally:
+        client.close()
+
+    stack.compose("down", "--remove-orphans", timeout=300)
+    stack.compose("up", "--detach", timeout=UP_TIMEOUT_SECONDS)
+    stack.wait_until_ready()
+
+    client = stack.client()
+    try:
+        after = {a["id"] for a in client.get("/inventory/assets").json()["items"]}
+    finally:
+        client.close()
+    assert after == before, "the same assets, from the same database, on the same volume"
+
+
+# ── operating it ─────────────────────────────────────────────────────────────
+
+
+@needs_compose
+def test_backup_verify_and_diagnostics_run_as_container_commands(stack):
+    backup = stack.compose("run", "--rm", "--no-deps", "api", "backup", "--reason", "compose")
+    manifest = json.loads(backup.stdout[backup.stdout.index("{") :])
+    assert manifest["name"].startswith("2")
+
+    verified = stack.compose(
+        "run", "--rm", "--no-deps", "api", "verify-backup", f"/data/backups/{manifest['name']}"
+    )
+    assert json.loads(verified.stdout[verified.stdout.index("{") :])["ok"] is True
+
+    report = stack.compose("run", "--rm", "--no-deps", "api", "diagnostics")
+    diagnostics = json.loads(report.stdout[report.stdout.index("{") :])
+    components = {c["name"]: c["path"] for c in diagnostics["components"]}
+    # Database, WAL, thumbnail cache, and backups all live in the mounted volume.
+    for name in ("database", "write_ahead_log", "thumbnail_cache", "backups"):
+        assert components[name].startswith("/data/"), (name, components[name])
+    assert {tool["name"] for tool in diagnostics["tools"]} >= {"exiftool", "immich-go"}
+    # And the worker running beside this command is visible as the lock's holder.
+    assert diagnostics["locks"]["worker"]["role"] == "worker"
+
+
+if __name__ == "__main__":  # a quick way to run just this file
+    raise SystemExit(pytest.main([__file__, "-v", *sys.argv[1:]]))

tests/integration/test_compose_runtime.py

@@ -0,0 +1,299 @@
+"""US08-03: the composition's contract, and the library-root check it depends on.
+
+Bringing the stack up needs a Docker daemon and the network, which is what
+``tests/e2e/test_compose_stack.py`` does. What can be checked without either is
+checked here, because the parts that rot silently — a second writer that is only
+prevented by convention, a data volume that stopped being the same volume for both
+roles, migrations that stopped running first, a committed value in a file that must
+carry none — are all readable from the files.
+
+The startup refusal is the other half: in a container the configured library roots
+must name the mount paths, and a mismatch has to fail before the lock is taken, not
+at the first rename.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import stat
+from pathlib import Path
+
+import pytest
+import yaml
+
+from photo_pipeline import path_policy
+from photo_pipeline.__main__ import main
+from photo_pipeline.config import Config
+
+REPO = Path(__file__).resolve().parents[2]
+COMPOSE_FILE = REPO / "docker-compose.yml"
+COMPOSE = yaml.safe_load(COMPOSE_FILE.read_text())
+ENV_EXAMPLE = REPO / ".env.example"
+
+SERVICES = COMPOSE["services"]
+DATA_VOLUME = "data:/data"
+
+
+def env_example_keys() -> list[str]:
+    """The variables the example file declares, in file order."""
+    return [
+        line.split("=", 1)[0]
+        for line in ENV_EXAMPLE.read_text().splitlines()
+        if "=" in line and not line.lstrip().startswith("#")
+    ]
+
+
+# ── one API, one worker, one library, one volume ─────────────────────────────
+
+
+def test_exactly_one_serving_and_one_working_container_from_the_same_image():
+    roles = {name: service["command"][0] for name, service in SERVICES.items()}
+    assert sorted(roles.values()) == ["migrate", "serve", "worker"]
+    assert [name for name, role in roles.items() if role == "serve"] == ["api"]
+    assert [name for name, role in roles.items() if role == "worker"] == ["worker"]
+
+    images = {service["image"] for service in SERVICES.values()}
+    assert len(images) == 1, "both roles must run the same build of the application"
+    assert "latest" not in images.pop()
+    # No `replicas`/`scale` key promising a second worker is fine; the lock decides.
+    assert not any("deploy" in service for service in SERVICES.values())
+
+
+def test_both_roles_share_the_data_volume_so_the_lock_is_visible_to_both():
+    """A second worker is refused by the library lock (US07-05) only if it can see it."""
+    for name, service in SERVICES.items():
+        assert DATA_VOLUME in service["volumes"], name
+    assert COMPOSE["volumes"]["data"]["driver"] == "local"
+    text = COMPOSE_FILE.read_text()
+    # The composition has to say why, because the failure is silent corruption.
+    assert "WAL" in text and re.search(r"NFS|SMB|network", text)
+
+
+def test_the_library_is_a_bind_mount_whose_target_is_the_configured_root():
+    for name, service in SERVICES.items():
+        mounts = [volume for volume in service["volumes"] if volume != DATA_VOLUME]
+        assert len(mounts) == 1, name
+        source, target = re.match(r"^(\$\{.*?\}):(\$\{.*?\})$", mounts[0]).groups()
+        # An unset host path fails the composition rather than mounting something else.
+        assert source.startswith("${PHOTO_PIPELINE_LIBRARY_HOST_PATH:?")
+        # The container-side path and the configured root are one variable, so they
+        # cannot drift apart into a library that is mounted but not configured.
+        assert target.startswith("${PHOTO_PIPELINE_LIBRARY_ROOTS:?")
+        assert service["environment"]["PHOTO_PIPELINE_LIBRARY_ROOTS"] == (
+            "${PHOTO_PIPELINE_LIBRARY_ROOTS}"
+        )
+        assert service["environment"]["PHOTO_PIPELINE_DATA_DIR"] == "/data"
+
+
+def test_migrations_run_to_completion_before_either_role_accepts_work():
+    """`migrate` runs the backup-then-migrate path, and a failed upgrade exits
+    non-zero with its pre-migration backup intact — proven in
+    tests/integration/test_backup_recovery.py. What the composition adds is that
+    neither role starts until it succeeded."""
+    assert SERVICES["migrate"]["command"] == ["migrate"]
+    assert SERVICES["migrate"]["restart"] == "no", "a one-shot that retries is not a gate"
+    for role in ("api", "worker"):
+        assert SERVICES[role]["depends_on"] == {
+            "migrate": {"condition": "service_completed_successfully"}
+        }, role
+
+
+def test_the_api_port_is_published_to_host_loopback_by_default():
+    published = SERVICES["api"]["ports"]
+    assert published == [
+        "${PHOTO_PIPELINE_PUBLISH_ADDRESS:-127.0.0.1}:${PHOTO_PIPELINE_PORT:-8000}:8000"
+    ]
+    # Reachable from the host means reachable from elsewhere as far as the app is
+    # concerned, so the access secret stays mandatory (US08-01).
+    assert SERVICES["api"]["environment"]["PHOTO_PIPELINE_HOST"] == "0.0.0.0"
+    assert SERVICES["api"]["environment"]["PHOTO_PIPELINE_PORT"] == 8000
+    assert "PHOTO_PIPELINE_ACCESS_SECRET" not in SERVICES["api"]["environment"]
+
+
+def test_containers_restart_by_themselves_and_stop_with_time_to_drain():
+    for role in ("api", "worker"):
+        assert SERVICES[role]["restart"] == "unless-stopped", role
+        assert SERVICES[role]["stop_grace_period"] == "30s", role
+
+
+def test_the_containers_run_as_the_library_owner_and_never_as_root():
+    for name, service in SERVICES.items():
+        assert service["user"] == "${PHOTO_PIPELINE_UID:-1000}:${PHOTO_PIPELINE_GID:-1000}", name
+        assert service["build"]["args"]["UID"] == "${PHOTO_PIPELINE_UID:-1000}", name
+
+
+# ── configuration comes from the environment, never from a committed file ────
+
+
+def test_configuration_and_secrets_come_from_the_environment_only():
+    for name, service in SERVICES.items():
+        assert service["env_file"] == ["${PHOTO_PIPELINE_ENV_FILE:-.env}"], name
+        for key, value in service["environment"].items():
+            # Every value is either a variable reference or a property of the
+            # composition itself (the volume path, the container's own port).
+            composed = isinstance(value, int) or value in ("/data", "0.0.0.0")
+            assert composed or value.startswith("${"), (name, key, value)
+    assert not (REPO / ".env").is_file() or ".env" in (REPO / ".gitignore").read_text()
+
+
+def test_the_example_file_lists_every_setting_and_carries_no_values():
+    declared = env_example_keys()
+    assert declared == sorted(set(declared), key=declared.index), "no variable twice"
+    for line in ENV_EXAMPLE.read_text().splitlines():
+        if "=" in line and not line.lstrip().startswith("#"):
+            assert line.endswith("="), f"a value in the example file: {line}"
+
+    expected = {f"PHOTO_PIPELINE_{name.upper()}" for name in Config.model_fields}
+    assert expected <= set(declared), sorted(expected - set(declared))
+    # And every variable the composition substitutes is documented there too.
+    substituted = set(re.findall(r"\$\{(PHOTO_PIPELINE_[A-Z_]+)", COMPOSE_FILE.read_text()))
+    assert substituted <= set(declared), sorted(substituted - set(declared))
+
+
+def test_the_example_file_is_not_a_dotenv_that_could_be_loaded_by_accident():
+    """`.env.example` must not be what `.env` is: no values means nothing to leak."""
+    assert ENV_EXAMPLE.name != ".env"
+    parsed = {k: v for k, v in _parse(ENV_EXAMPLE.read_text()).items() if v}
+    assert parsed == {}
+
+
+def _parse(text: str) -> dict[str, str]:
+    from photo_pipeline.config import parse_env_file
+
+    return parse_env_file(text)
+
+
+# ── the lock across container lifetimes ──────────────────────────────────────
+
+
+def test_a_lock_left_by_a_container_that_is_gone_does_not_block_the_restart(
+    tmp_path, monkeypatch
+):
+    """A restarted container is a new hostname and a recycled pid 1, so the record
+    in the lock file proves nothing; the kernel's flock does (US08-03)."""
+    from photo_pipeline.services.app_lock import LibraryLock
+
+    config = Config(data_dir=tmp_path / "data", library_roots=(tmp_path,))
+    (tmp_path / "data").mkdir()
+    (tmp_path / "data" / "worker.lock.json").write_text(
+        json.dumps(
+            {
+                "lock_version": 1,
+                "role": "worker",
+                "pid": 1,  # pid 1 of a container that no longer exists
+                "host": "3f2a1b9c4d5e",  # its hostname was its container id
+                "started_at": "2026-01-01T00:00:00+00:00",
+                "library_roots": ["/library"],
+            }
+        )
+    )
+    monkeypatch.setattr(path_policy, "in_container", lambda: True)
+
+    taken = LibraryLock(config, "worker").acquire()
+
+    assert taken.pid == os.getpid(), "the worker must come back after a restart"
+
+
+def test_a_second_worker_is_still_refused_while_the_first_holds_the_lock(tmp_path, monkeypatch):
+    """The other half: the same flock refuses a concurrent second writer, whether it
+    is a process or another container of the same composition."""
+    from photo_pipeline.services.app_lock import LibraryLock, LockHeld
+
+    config = Config(data_dir=tmp_path / "data", library_roots=(tmp_path,))
+    monkeypatch.setattr(path_policy, "in_container", lambda: True)
+    first = LibraryLock(config, "worker")
+    first.acquire()
+
+    with pytest.raises(LockHeld, match="worker is already running"):
+        LibraryLock(config, "worker").acquire()
+
+    first.release()
+    LibraryLock(config, "worker").acquire()  # free again
+
+
+# ── the startup check the mount depends on ───────────────────────────────────
+
+
+def test_configured_roots_that_are_mounted_and_writable_are_accepted(tmp_path):
+    assert path_policy.roots_refusal([tmp_path]) is None
+    assert path_policy.roots_refusal([]) is None, "no roots is a configuration, not a fault"
+
+
+def test_an_unmounted_library_root_is_refused_by_name(tmp_path):
+    refusal = path_policy.roots_refusal([tmp_path / "srv" / "photos"])
+    assert refusal is not None
+    assert "does not exist" in refusal and "PHOTO_PIPELINE_LIBRARY_ROOTS" in refusal
+
+
+def test_a_root_that_is_not_a_directory_or_not_readable_is_refused(tmp_path):
+    a_file = tmp_path / "photos.txt"
+    a_file.write_text("not a library")
+    assert "not a directory" in path_policy.roots_refusal([a_file])
+
+    unreadable = tmp_path / "unreadable"
+    unreadable.mkdir()
+    unreadable.chmod(0o000)
+    try:
+        refusal = path_policy.roots_refusal([unreadable])
+    finally:
+        unreadable.chmod(0o755)
+    if os.getuid() != 0:  # root ignores the mode, and CI may well be root
+        assert refusal is not None and "not readable" in refusal
+
+
+def test_an_unwritable_root_is_not_refused_here(tmp_path):
+    """A bind mount's ownership is virtualised on macOS and Windows, so os.access
+    would refuse a working deployment. The real errno at the first rename is at
+    least true; this check is about the mount, not the mode."""
+    read_only = tmp_path / "read-only"
+    read_only.mkdir()
+    read_only.chmod(stat.S_IRUSR | stat.S_IXUSR)
+    try:
+        assert path_policy.roots_refusal([read_only]) is None
+    finally:
+        read_only.chmod(0o755)
+
+
+def test_in_a_container_a_root_that_was_never_mounted_is_refused(tmp_path, monkeypatch):
+    """The container-only failure: the path exists, but it belongs to the image."""
+    unmounted = tmp_path / "library"
+    (unmounted / "album").mkdir(parents=True)
+    refusal = path_policy.roots_refusal([unmounted], require_mount=True)
+    assert refusal is not None and "not on a mounted filesystem" in refusal
+
+    # A bind mount is a mount point, and a root *below* one is mounted too: a
+    # deployment may mount /srv and configure /srv/photos.
+    monkeypatch.setattr(os.path, "ismount", lambda path: Path(path) == unmounted.resolve())
+    assert path_policy.roots_refusal([unmounted], require_mount=True) is None
+    assert path_policy.roots_refusal([unmounted / "album"], require_mount=True) is None
+
+
+@pytest.mark.parametrize("role", ["serve", "worker"])
+def test_a_root_mismatch_refuses_at_startup_before_any_lock_is_taken(
+    role, tmp_path, monkeypatch, capsys
+):
+    data = tmp_path / "data"
+    monkeypatch.setenv("PHOTO_PIPELINE_DATA_DIR", str(data))
+    monkeypatch.setenv("PHOTO_PIPELINE_LIBRARY_ROOTS", str(tmp_path / "not-mounted"))
+    monkeypatch.setattr(path_policy, "in_container", lambda: True)
+
+    assert main([role]) == 5
+    assert "does not exist" in capsys.readouterr().err
+    assert not list(data.glob("*.lock.json")), "nothing started, so nothing is locked"
+
+
+@pytest.mark.parametrize("role", ["serve", "worker"])
+def test_an_unmounted_root_on_a_host_is_not_a_reason_to_refuse(role, tmp_path, monkeypatch):
+    """An archive medium that is not plugged in is a Tuesday, not a misconfiguration:
+    refusing would take the offline half of the library away with it (concept §9)."""
+    monkeypatch.setenv("PHOTO_PIPELINE_DATA_DIR", str(tmp_path / "data"))
+    monkeypatch.setenv("PHOTO_PIPELINE_LIBRARY_ROOTS", str(tmp_path / "not-mounted"))
+    monkeypatch.setenv("PHOTO_PIPELINE_ACCESS_SECRET", "unused-on-loopback")
+    monkeypatch.setattr(path_policy, "in_container", lambda: False)
+    # Reaching the lock is the proof: that is the next thing either role does, and
+    # stopping there keeps the test out of a uvicorn/worker loop.
+    monkeypatch.setattr("photo_pipeline.__main__._acquire", lambda *_, **__: 99)
+
+    assert main([role]) == 99

tests/story_traceability.json

@@ -182,10 +182,13 @@
     "US08-02": [
       "tests/integration/test_container_image.py",
       "tests/e2e/test_container_runtime.py"
+    ],
+    "US08-03": [
+      "tests/integration/test_compose_runtime.py",
+      "tests/e2e/test_compose_stack.py"
     ]
   },
   "planned": [
-    "US08-03",
     "US08-04",
     "US08-05"
   ],