"""Phase H — the container acceptance gate (US08-05).

The deployed application is verified the way the host application is: real image,
real composition, real browser, real restarts. Everything below runs against
containers that this suite provisions from the built image, against a temporary
fixture library on a bind mount and an isolated data volume, and destroys afterwards.

Four journeys, one per thing a deployment can get wrong:

* **the browser journey** — discovery, duplicate review, analysis, album proposal,
  rename, upload preflight, and archive, driven through the containerized frontend;
* **the upgrade journey** — the previous version's image runs first, then this one,
  and the database, its migrations, the rename journal, the job history, and the
  thumbnail cache have to still be there;
* **the restart journey** — both containers are killed mid-job and the work resumes
  without doing anything twice;
* **the security gates** — the refusals a loopback deployment made are still made
  behind a published port: no session, forged forwarded headers, a path that leaves
  the mounted library, and a secret in the logs.

Run it as one command, with evidence: ``python -m photo_pipeline container-gate``.
"""

from __future__ import annotations

import collections
from contextlib import closing
import json
import os
import shutil
import subprocess
import sys
import time
from pathlib import Path

import httpx
import pytest
from playwright.sync_api import expect

from photo_pipeline.db import head_revision
from tests.e2e._container_harness import (
    Stack,
    await_job,
    build_image,
    compose_available,
    needs_compose,
    remove_library,
    temporary_library,
    write_env_file,
)

pytestmark = [pytest.mark.phase_h, pytest.mark.container]

PROJECT = "photo-pipeline-us0805"
IMAGE = "photo-pipeline-test:us08-05"
# The version being upgraded *from*. CI passes the tag it last published; without one,
# the previous commit's tree is built, which is the same claim without a registry.
PREVIOUS_IMAGE = os.environ.get("PHOTO_PIPELINE_PREVIOUS_IMAGE")
PREVIOUS_TAG = "photo-pipeline-test:us08-05-previous"
SECRET = "container-gate-access-secret"
IMMICH_SENTINEL = "immich-sentinel-9f3a2b"
HOSTNAME = "photos.test"
ALBUM = "rome"
RENAMED = "2019 Rome"
BURST = 30


# ── the image and the stacks ─────────────────────────────────────────────────


@pytest.fixture(scope="session")
def image() -> str:
    if not compose_available():
        pytest.skip("no Docker daemon with the compose plugin")
    return build_image(IMAGE)


@pytest.fixture(scope="module")
def journey(image, tmp_path_factory):
    """One album, one exact duplicate of a photo in it, and the exclusion sentinel."""
    library = temporary_library({ALBUM: 3}, prefix="us0805-journey-")
    shutil.copyfile(library / ALBUM / f"{ALBUM}_0.jpg", library / ALBUM / "copy.jpg")
    env_file = write_env_file(
        tmp_path_factory.mktemp("journey") / "gate.env",
        SECRET,
        {
            "PHOTO_PIPELINE_ALLOWED_HOSTS": HOSTNAME,
            # Configured but never reachable: the upload view's job here is to show the
            # preflight blockers, and the key's job is to be absent from every log.
            "PHOTO_PIPELINE_IMMICH_API_KEY": IMMICH_SENTINEL,
            "PHOTO_PIPELINE_IMMICH_SERVER_URL": "http://127.0.0.1:1",
        },
    )
    stack = Stack(library, env_file, project=f"{PROJECT}-journey", image=image, secret=SECRET)
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()
        _seed_journey(stack)
        yield stack
    finally:
        stack.down()
        remove_library(library)


def _seed_journey(stack: Stack) -> None:
    """Everything the views need, established over the public API before they render."""
    with closing(stack.client()) as client:
        client.post("/inventory/scan").raise_for_status()
        client.post("/duplicates/detect").raise_for_status()
        queue = client.get("/safety/queue", params={"limit": 100}).json()["items"]
        for item in queue:
            decided = client.post(
                "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
            )
            assert decided.status_code == 200, decided.text
        job = client.post("/analysis/jobs").json()
        await_job(client, job["id"])
        client.post("/albums/proposals", json={}).raise_for_status()
        # An archive destination inside the data volume: a second mount would prove
        # nothing more, and the view needs a location to have something to show.
        stack.compose("exec", "-T", "api", "mkdir", "-p", "/data/archive")
        client.post(
            "/archive-locations", json={"name": "external", "root": "/data/archive"}
        ).raise_for_status()


# ── the browser journey ──────────────────────────────────────────────────────


@needs_compose
def test_the_browser_journey_covers_every_stage_view_of_the_deployed_app(page, journey):
    """One pass through the deployed frontend, in workflow order.

    The access secret is supplied the way a person supplies it — the app asks, the
    answer is kept for the tab — so what is proven is the authenticated deployment,
    not a test-only bypass.
    """
    page.on("dialog", lambda dialog: dialog.accept(SECRET))
    errors: list[str] = []
    page.on("pageerror", lambda error: errors.append(str(error)))
    base = journey.base
    with closing(journey.client()) as client:
        cluster = client.get("/duplicates/clusters").json()["items"][0]

    # ── discovery ────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/workflow")
    page.get_by_test_id("stage-safety").wait_for()
    expect(page.get_by_test_id("stage-analysis")).to_be_visible()

    page.goto(f"{base}/app/#/inventory")
    rows = page.get_by_test_id("asset-row")
    rows.first.wait_for()
    assert rows.count() == 4, "three photos and the duplicate copy; never the sentinel"
    assert "sentinel" not in page.content() and "_IGNORE" not in page.content()

    # ── duplicate review ─────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/duplicates/{cluster['id']}")
    page.get_by_test_id("cluster-state").wait_for()
    # Exact bytes: the cluster arrives decided, and the review surface has to show
    # both members and the evidence the decision was made on.
    expect(page.get_by_test_id("cluster-state")).to_contain_text("decided")
    expect(page.get_by_test_id("member")).to_have_count(2)
    expect(page.get_by_test_id("member").first).to_contain_text("/library/")

    # ── analysis ─────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/analyze")
    page.get_by_test_id("analyze-counts").wait_for()
    expect(page.get_by_test_id("run-analysis")).to_be_visible()

    # ── album proposal ───────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/albums?album={ALBUM}")
    page.get_by_test_id("suggested-name").wait_for()
    page.get_by_test_id("final-name").fill(RENAMED)
    page.get_by_test_id("save-name").click()
    page.get_by_test_id("approve").click()
    expect(page.get_by_test_id("proposal-status")).to_contain_text("approved")

    # ── rename, applied against the bind mount ───────────────────────────────
    page.goto(f"{base}/app/#/renames")
    page.get_by_test_id("build-plan").click()
    page.get_by_test_id("operations").wait_for()
    expect(page.get_by_test_id("op-destination").first).to_contain_text(RENAMED)
    page.get_by_test_id("apply-plan").click()
    expect(page.get_by_test_id("apply-result")).to_contain_text("Applied 1, failed 0")
    # The mounted library is the host's directory: the container renamed the operator's
    # folder, not a copy inside its own layer.
    assert (journey.library / RENAMED).is_dir()
    assert not (journey.library / ALBUM).exists()

    # ── upload preflight ─────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/uploads")
    page.get_by_test_id("upload-scope").wait_for()
    expect(page.get_by_test_id("album-row").first).to_be_visible()
    assert IMMICH_SENTINEL not in page.content(), "the API key never reaches the browser"

    # ── archive ──────────────────────────────────────────────────────────────
    page.goto(f"{base}/app/#/archive")
    page.get_by_test_id("archive-locations").wait_for()
    expect(page.get_by_test_id("location-row").first).to_contain_text("external")

    assert errors == [], f"the deployed frontend raised page errors: {errors}"


# ── the upgrade journey ──────────────────────────────────────────────────────


@pytest.fixture(scope="module")
def previous_image(image) -> str:
    """The image the deployment is upgrading *from*.

    The published tag when there is one. Before the first publish there is nothing to
    pull, and refusing then would mean the gate could never let the first deploy
    through — so the previous commit's tree is built instead, which is the same claim
    without a registry.
    """
    if PREVIOUS_IMAGE:
        pulled = subprocess.run(
            ["docker", "pull", PREVIOUS_IMAGE], capture_output=True, timeout=1800
        )
        if pulled.returncode == 0:
            return PREVIOUS_IMAGE
    return build_image(PREVIOUS_TAG, revision="HEAD~1")


@needs_compose
def test_an_upgrade_keeps_the_database_the_journal_the_jobs_and_the_cache(
    image, previous_image, tmp_path_factory
):
    library = temporary_library({ALBUM: 2}, prefix="us0805-upgrade-")
    env_file = write_env_file(tmp_path_factory.mktemp("upgrade") / "gate.env", SECRET)
    stack = Stack(
        library, env_file, project=f"{PROJECT}-upgrade", image=previous_image, secret=SECRET
    )
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()

        # ── what the previous version leaves behind ──────────────────────────
        with closing(stack.client()) as client:
            client.post("/inventory/scan").raise_for_status()
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            for item in client.get("/safety/queue", params={"limit": 100}).json()["items"]:
                client.post(
                    "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
                ).raise_for_status()
            job = client.post("/analysis/jobs").json()
            finished = await_job(client, job["id"])
            for asset in assets:  # populate the thumbnail cache
                thumbnail = client.get(f"/assets/{asset['id']}/thumbnail", params={"size": 256})
                assert thumbnail.status_code == 200, thumbnail.text
            # A rename plan, left unapplied: the journal has to survive the upgrade
            # exactly as it was, or a half-applied one could not be recovered.
            client.post("/albums/proposals", json={}).raise_for_status()
            _approve(client, RENAMED)
            plan = client.post("/rename-plans").json()

        before = {
            "assets": sorted(asset["id"] for asset in assets),
            "job": finished["progress"],
            "plan": (plan["id"], plan["checksum"]),
            "thumbnails": _cache_files(stack),
            "revision": _revision(stack),
        }
        assert before["thumbnails"], "no thumbnail was cached, so nothing would be proven"

        # ── the upgrade: same volume, new image ──────────────────────────────
        stack.down(volumes=False)
        stack.use_image(image)
        stack.up()
        stack.wait_until_ready()

        after_revision = _revision(stack)
        assert after_revision == head_revision(), "the new image did not migrate the volume"
        if after_revision != before["revision"]:
            # A schema change is snapshotted before it is applied (US07-05), so a
            # failed upgrade is restorable rather than a lost library.
            assert _ls(stack, "/data/backups"), "a migration ran without a backup"

        with closing(stack.client()) as client:
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            assert sorted(asset["id"] for asset in assets) == before["assets"]
            assert client.get(f"/jobs/{job['id']}").json()["progress"] == before["job"]
            plans = client.get("/rename-plans").json()["items"]
            assert (plans[0]["id"], plans[0]["checksum"]) == before["plan"]
            for asset in assets:
                assert (
                    client.get(f"/analysis/results/{asset['id']}").status_code == 200
                ), "an analysis result did not survive the upgrade"
        # The cache is keyed by pixel hash and thumbnail version, so an upgrade that
        # kept the volume must keep the files: regenerating them is work nobody asked
        # for, and losing them silently is how a cache stops being one.
        assert set(before["thumbnails"]) <= set(_cache_files(stack))
    finally:
        stack.down()
        remove_library(library)


def _approve(client: httpx.Client, name: str, *, album: str = ALBUM) -> None:
    for payload, route in (({"name": name}, "edit"), ({}, "approve")):
        current = client.get(f"/albums/proposals/{album}").json()
        client.post(
            f"/albums/proposals/{album}/{route}",
            json={**payload, "expected_version": current["version"]},
        ).raise_for_status()


def _exec(stack: Stack, *args: str) -> str:
    return stack.compose("exec", "-T", "api", *args).stdout


def _ls(stack: Stack, directory: str) -> list[str]:
    listing = stack.compose("exec", "-T", "api", "ls", directory, check=False)
    return [line for line in listing.stdout.split() if line]


def _cache_files(stack: Stack) -> list[str]:
    return sorted(
        _exec(stack, "find", "/data/cache", "-type", "f", "-name", "*.webp").split()
    )


def _revision(stack: Stack) -> str:
    """The schema revision the volume's database is actually at."""
    return _exec(
        stack,
        "python",
        "-c",
        "import sqlite3;print(sqlite3.connect('/data/photo_pipeline.db')"
        ".execute('select version_num from alembic_version').fetchone()[0])",
    ).strip()


# ── the restart journey ──────────────────────────────────────────────────────


@needs_compose
def test_killing_both_containers_mid_job_resumes_without_doing_anything_twice(
    image, tmp_path_factory
):
    """`docker kill` is the honest restart: no grace period, no orderly stop, no
    chance for either process to write a tidy final state."""
    library = temporary_library({"burst": BURST}, prefix="us0805-restart-")
    env_file = write_env_file(tmp_path_factory.mktemp("restart") / "gate.env", SECRET)
    stack = Stack(library, env_file, project=f"{PROJECT}-restart", image=image, secret=SECRET)
    try:
        stack.down()
        stack.up()
        stack.wait_until_ready()

        with closing(stack.client()) as client:
            client.post("/inventory/scan").raise_for_status()
            assets = client.get("/inventory/assets", params={"limit": 200}).json()["items"]
            assert len(assets) == BURST
            for item in client.get("/safety/queue", params={"limit": 100}).json()["items"]:
                client.post(
                    "/safety/decisions", json={"asset_id": item["asset_id"], "decision": "sfw"}
                ).raise_for_status()
            job = client.post("/analysis/jobs").json()
            progress = _wait_for_progress(client, job["id"])

        assert 0 < progress["done"] < progress["total"], progress
        stack.compose("kill", "api", "worker")

        stack.up()
        stack.wait_until_ready()
        with closing(stack.client()) as client:
            finished = await_job(client, job["id"])
            assert finished["progress"]["done"] == BURST, finished
            results = [
                client.get(f"/analysis/results/{asset['id']}").json() for asset in assets
            ]

        # Exactly one stored result per asset: at-least-once execution, idempotent
        # recovery — a retried item overwrites its own attempt, it does not add one.
        assert len(results) == BURST
        assert all(result["description"] for result in results)
        # And the side effect nobody can take back — the call to the provider — happened
        # again only for whatever was in flight when the containers died.
        analysed = collections.Counter(_exec(stack, "cat", "/data/vision.log").split())
        assert len(analysed) == BURST, "every photo was analysed, and only the library's"
        assert max(analysed.values()) <= 2, dict(analysed)
        assert sum(1 for count in analysed.values() if count > 1) <= 1, dict(analysed)
    finally:
        stack.down()
        remove_library(library)


def _wait_for_progress(client: httpx.Client, job_id: str, *, timeout: float = 120) -> dict:
    """Wait until the job is provably under way but provably unfinished."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        snapshot = client.get(f"/jobs/{job_id}").json()
        progress = snapshot["progress"]
        if progress["done"] and progress["done"] < progress["total"]:
            return progress
        if snapshot["state"] in ("succeeded", "failed"):
            raise AssertionError(f"the job finished before it could be interrupted: {snapshot}")
        time.sleep(0.05)
    raise AssertionError(f"the job never started: {client.get(f'/jobs/{job_id}').json()}")


# ── the security gates ───────────────────────────────────────────────────────


@needs_compose
def test_the_deployed_instance_refuses_a_caller_without_a_session(journey):
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        for method, path in (("GET", "/workflow"), ("POST", "/inventory/scan")):
            response = client.request(method, path)
            assert response.status_code == 401, path
            assert response.json()["error"]["code"] == "unauthenticated"
        # A session still has to be paid for with the operator's secret.
        assert client.get("/session", headers={"X-Access-Secret": "guessed"}).status_code == 401
        # Readiness stays open: the orchestrator's health check holds no session.
        assert client.get("/health/ready").status_code == 200


@needs_compose
def test_forged_forwarded_headers_cannot_smuggle_an_allowed_host_past_the_check(journey):
    """Behind a proxy the app believes ``X-Forwarded-*`` — but only from the proxy.
    Nothing in this composition is a trusted proxy, so the claim is the client's."""
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        client.headers["X-CSRF-Token"] = (
            client.get("/session", headers={"X-Access-Secret": SECRET}).json()["csrf_token"]
        )
        # The hostname this deployment is reached under is accepted.
        assert client.get("/workflow", headers={"Host": HOSTNAME}).status_code == 200

        forged = client.get(
            "/workflow",
            headers={"Host": "photos.evil.example", "X-Forwarded-Host": HOSTNAME},
        )
        assert forged.status_code == 403
        assert forged.json()["error"]["code"] == "host_not_allowed"
        # A forged protocol claim must not mark the session cookie as HTTPS-only
        # either — that would strand the operator's real, plain-HTTP session.
        bootstrap = httpx.get(
            f"{journey.base}/api/v1/session",
            headers={"X-Access-Secret": SECRET, "X-Forwarded-Proto": "https"},
            timeout=30,
        )
        assert "secure" not in bootstrap.headers["set-cookie"].lower()


@needs_compose
def test_a_path_that_leaves_the_mounted_library_is_refused(journey):
    """The container's own filesystem is not the library. A symlink swapped under a
    known asset is the sharpest version of the question, because the database still
    points at a path inside the mount."""
    with closing(journey.client()) as client:
        asset = client.get("/inventory/assets", params={"limit": 200}).json()["items"][0]
        original = journey.library / Path(asset["current_path"]).relative_to("/library")
        kept = original.read_bytes()
        original.unlink()
        original.symlink_to("/etc/passwd")
        try:
            escaped = client.get(f"/assets/{asset['id']}/thumbnail", params={"size": 256})
        finally:
            original.unlink()
            original.write_bytes(kept)

        assert escaped.status_code == 403
        assert escaped.json()["error"]["code"] == "path_not_allowed"
        assert "root:" not in escaped.text

        # And a root that names nothing mounted is refused before the process serves.
        refused = journey.compose(
            "run",
            "--rm",
            "--no-deps",
            "--env",
            "PHOTO_PIPELINE_LIBRARY_ROOTS=/srv/photos",
            "api",
            "serve",
            check=False,
        )
        assert refused.returncode == 5, refused.stdout + refused.stderr


@needs_compose
def test_no_secret_reaches_the_container_logs(journey):
    with httpx.Client(base_url=f"{journey.base}/api/v1", timeout=30) as client:
        client.get("/session", headers={"X-Access-Secret": "wrong-secret-attempt"})
        client.get("/session", headers={"X-Access-Secret": SECRET})
    logs = journey.logs()
    assert SECRET not in logs
    assert IMMICH_SENTINEL not in logs
    assert "wrong-secret-attempt" not in logs, "a rejected secret is still a secret"
    # The refusal itself is logged, so an operator can see the attempt.
    assert "access secret rejected" in logs


@needs_compose
def test_the_evidence_of_this_run_names_the_stack_it_was_produced_from(journey):
    """A gate that cannot say what it ran against is an opinion. `docker compose ps`
    is the record: one API, one worker, one completed migration."""
    listing = [
        json.loads(line)
        for line in journey.compose("ps", "--all", "--format", "json").stdout.splitlines()
        if line.strip()
    ]
    services = {entry["Service"]: entry["State"] for entry in listing}
    assert services["api"] == "running" and services["worker"] == "running"
    assert services["migrate"] == "exited"


if __name__ == "__main__":  # a quick way to run just this file
    raise SystemExit(pytest.main([__file__, "-v", *sys.argv[1:]]))