photoanalyzer/tests/e2e/test_compose_stack.py

"""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).

The stack itself lives in ``tests/e2e/_container_harness.py``, shared with the
container acceptance gate (US08-05).
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

import pytest

from tests.e2e._container_harness import (
    CONTAINER_LIBRARY,
    Stack,
    await_job,
    compose_available,
    needs_compose,
    remove_library,
    temporary_library,
    write_env_file,
)

PROJECT = "photo-pipeline-us0803"
IMAGE = "photo-pipeline-test:us08-03"
SECRET = "compose-acceptance-secret"

pytestmark = pytest.mark.container


@pytest.fixture(scope="module")
def library() -> Path:
    root = temporary_library({"01_day": 2, "02_night": 1})
    try:
        yield root
    finally:
        remove_library(root)


@pytest.fixture(scope="module")
def env_file(tmp_path_factory) -> Path:
    return write_env_file(tmp_path_factory.mktemp("config") / "compose.env", SECRET)


@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, project=PROJECT, image=IMAGE, secret=SECRET)
    running.down()
    running.up("--build")
    try:
        running.wait_until_ready()
        yield running
    finally:
        running.down()


# ── 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.down(volumes=False)
    stack.up()
    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:]]))