photoanalyzer/tests/integration/test_upload_batches.py

"""Upload batch orchestration (US05-02).

The uploader is a real executable on disk driven through the real integration
adapter and ``subprocess`` — never a mock — so argument construction, output
bounding, credential privacy, and killing a running process are exercised the way
production does them. Each fake uploader records the argv it was given into a
side-channel file, which is what the argument/album-isolation assertions read.
"""

import json
import os
import signal
import stat
import threading
import time
import uuid
from datetime import datetime, timezone
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path

import pytest
from fastapi.testclient import TestClient

from photo_pipeline.api.app import create_app
from photo_pipeline.config import Config
from photo_pipeline.db import create_db_engine, create_session_factory, run_migrations
from photo_pipeline.jobs.domain_handlers import UPLOAD_BATCH
from photo_pipeline.jobs.worker import Worker
from photo_pipeline.models import AnalysisResult, Asset, SafetyReview, UploadBatch
from photo_pipeline.services.hashing import sha1_file, sha256_file
from photo_pipeline.services.jobs import JobService, JobState
from photo_pipeline.services.upload_batches import (
    BatchConflict,
    BatchError,
    BatchState,
    ItemState,
    UploadBatchService,
)
from photo_pipeline.services.uploads import UploadService

pytestmark = pytest.mark.phase_e  # part of the Phase E acceptance gate (US05-06)

NOW = datetime(2026, 1, 1, tzinfo=timezone.utc)
SENTINEL_KEY = "immich-sentinel-9f3a2b"
UPLOADER_VERSION = "immich-go 0.21.0"


# ── fake external boundary ───────────────────────────────────────────────────


class _PingHandler(BaseHTTPRequestHandler):
    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler API)
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(b'{"res":"pong"}')

    def log_message(self, *args):
        pass


@pytest.fixture
def immich_server():
    server = HTTPServer(("127.0.0.1", 0), _PingHandler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_port}"
    server.shutdown()
    server.server_close()


def _uploader(tmp_path, body: str = "", *, name="immich-go"):
    """A real executable standing in for immich-go.

    ``--version`` answers like the real tool; any other invocation appends its
    complete argv to ``<name>.argv`` and then runs ``body``.
    """
    path = tmp_path / name
    argv_log = tmp_path / f"{name}.argv"
    path.write_text(
        "#!/bin/sh\n"
        f'if [ "$1" = "--version" ]; then echo "{UPLOADER_VERSION}"; exit 0; fi\n'
        f'printf "%s\\n" "$*" >> "{argv_log}"\n'
        f"{body}\n"
    )
    path.chmod(path.stat().st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)
    return path


def _argv(tmp_path, name="immich-go") -> list[str]:
    log = tmp_path / f"{name}.argv"
    return log.read_text().splitlines() if log.exists() else []


# ── environment ──────────────────────────────────────────────────────────────


def _env(tmp_path, server_url, *, uploader=None):
    (tmp_path / "data").mkdir(exist_ok=True)
    lib = tmp_path / "lib"
    lib.mkdir(exist_ok=True)
    config = Config.from_env(
        {
            "PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data"),
            "PHOTO_PIPELINE_LIBRARY_ROOTS": str(lib),
            "PHOTO_PIPELINE_IMMICH_SERVER_URL": server_url,
            "PHOTO_PIPELINE_IMMICH_API_KEY": SENTINEL_KEY,
            "PHOTO_PIPELINE_IMMICH_GO_BINARY": str(
                uploader if uploader is not None else _uploader(tmp_path, "exit 0")
            ),
        }
    )
    run_migrations(config.database_url)
    return config, create_session_factory(create_db_engine(config.database_url)), lib


def _album(sf, lib, album="rome", names=("a.jpg", "b.jpg")):
    folder = lib / album
    folder.mkdir(parents=True, exist_ok=True)
    with sf() as session:
        for name in names:
            path = folder / name
            path.write_bytes(f"{album}/{name}".encode() * 16)
            asset_id = str(uuid.uuid4())
            session.add(
                Asset(
                    id=asset_id,
                    original_path=str(path),
                    current_path=str(path),
                    discovered_at=NOW,
                    hash_version=1,
                    byte_size=path.stat().st_size,
                    current_sha256=sha256_file(path),
                )
            )
            session.add(
                SafetyReview(
                    id=str(uuid.uuid4()),
                    asset_id=asset_id,
                    decision="sfw",
                    exif_verified_at=NOW,
                )
            )
            session.add(
                AnalysisResult(asset_id=asset_id, status="analyzed", exif_written_at=NOW)
            )
        session.commit()
    return folder


def _service(sf, config):
    return UploadBatchService(sf, config=config)


def _approved(sf, config, albums=None, **kwargs):
    """Create batches from a fresh, ready preflight."""
    report = UploadService(sf, config=config).preflight(albums, **kwargs)
    assert report["state"] == "ready", report["blockers"]
    return _service(sf, config).create(albums, token=report["token"], **kwargs)


# ── creation ─────────────────────────────────────────────────────────────────


def test_batch_records_album_assets_hashes_and_command(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)

    (batch,) = _approved(sf, config)

    assert batch["state"] == BatchState.PLANNED
    assert batch["album"] == "rome" and batch["folder"] == str(folder)
    assert batch["asset_count"] == 2 and len(batch["items"]) == 2
    assert batch["uploader_version"] == UPLOADER_VERSION
    assert "--album-name=rome" in batch["command"]
    assert "--api-key=***" in batch["command"]
    for item in batch["items"]:
        assert item["sha256"] == sha256_file(item["path"])
        assert item["sha1"] == sha1_file(item["path"])
        assert item["state"] == ItemState.PENDING


def test_one_batch_per_album_and_scope_is_isolated(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))

    batches = _approved(sf, config, ["paris"])

    assert [b["album"] for b in batches] == ["paris"]
    assert [b["asset_count"] for b in batches] == [1]


def test_creating_twice_reuses_the_open_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)

    first = _approved(sf, config)[0]
    second = _approved(sf, config)[0]

    assert first["id"] == second["id"]


def test_stale_preflight_token_cannot_create_a_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    token = UploadService(sf, config=config).preflight()["token"]
    (folder / "a.jpg").write_bytes(b"edited after approval")

    with pytest.raises(BatchConflict) as error:
        _service(sf, config).create(token=token)

    assert error.value.code == "stale_preflight"


def test_blocked_scope_cannot_create_a_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (folder / "a.jpg").write_bytes(b"edited before approval")
    report = UploadService(sf, config=config).preflight()

    with pytest.raises(BatchError) as error:
        _service(sf, config).create(token=report["token"])

    assert error.value.code == "not_ready"


# ── running the uploader ─────────────────────────────────────────────────────


def test_run_invokes_the_uploader_with_the_batch_album_and_folder(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (batch,) = _approved(sf, config)

    result = _service(sf, config).run(batch["id"])

    assert result["state"] == BatchState.SUCCEEDED and result["exit_code"] == 0
    (invocation,) = _argv(tmp_path)
    assert "upload from-folder" in invocation
    assert "--album-name=rome" in invocation
    assert str(folder) in invocation
    assert all(item["state"] == ItemState.SENT for item in result["items"])
    assert result["attempt_count"] == 1 and result["started_at"] and result["finished_at"]


def test_other_albums_are_never_passed_to_the_uploader(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    paris = _album(sf, lib, "paris", names=("c.jpg",))
    (batch,) = _approved(sf, config, ["rome"])

    _service(sf, config).run(batch["id"])

    (invocation,) = _argv(tmp_path)
    assert str(paris) not in invocation and "paris" not in invocation


def test_album_names_are_arguments_not_shell_text(tmp_path, immich_server):
    """A folder whose name contains shell metacharacters must reach the uploader
    verbatim; nothing may be interpreted (no shell is involved)."""
    config, sf, lib = _env(tmp_path, immich_server)
    hostile = "rome; touch pwned"
    _album(sf, lib, hostile, names=("a.jpg",))
    (batch,) = _approved(sf, config)

    result = _service(sf, config).run(batch["id"])

    assert result["state"] == BatchState.SUCCEEDED
    assert f"--album-name={hostile}" in _argv(tmp_path)[0]
    assert not (Path.cwd() / "pwned").exists() and not (lib / "pwned").exists()


def test_uploader_failure_is_recorded_and_items_are_not_sent(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, "echo 'boom' >&2; exit 3")
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)

    result = _service(sf, config).run(batch["id"])

    assert result["state"] == BatchState.FAILED and result["exit_code"] == 3
    assert result["error_code"] == "uploader_failed"
    assert all(item["state"] == ItemState.FAILED for item in result["items"])
    assert "boom" in _service(sf, config).report(batch["id"])


def test_a_failed_batch_can_be_retried_as_a_new_attempt(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server, uploader=_uploader(tmp_path, "exit 3"))
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    service.run(batch["id"])

    retried = service.run(batch["id"])

    assert retried["attempt_count"] == 2
    assert len(_argv(tmp_path)) == 2


def test_bytes_edited_after_approval_block_the_upload(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    folder = _album(sf, lib)
    (batch,) = _approved(sf, config)
    (folder / "a.jpg").write_bytes(b"edited after approval")

    with pytest.raises(BatchConflict) as error:
        _service(sf, config).run(batch["id"])

    assert error.value.code == "stale_preflight"
    assert _argv(tmp_path) == [], "the uploader must not run on changed bytes"
    assert _service(sf, config).get(batch["id"])["state"] == BatchState.FAILED


# ── credential privacy ───────────────────────────────────────────────────────


def test_the_api_key_reaches_the_uploader_but_never_the_record(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)

    result = _service(sf, config).run(batch["id"])

    assert f"--api-key={SENTINEL_KEY}" in _argv(tmp_path)[0], "the real key must be passed"
    assert SENTINEL_KEY not in json.dumps(result, default=str)


def test_a_key_echoed_by_the_uploader_is_scrubbed_from_the_report(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo "using key $4"; exit 0')
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)

    service.run(batch["id"])

    report = service.report(batch["id"])
    assert SENTINEL_KEY not in report and "***" in report


# ── bounded output ───────────────────────────────────────────────────────────


def test_a_chatty_uploader_cannot_grow_the_report_without_bound(tmp_path, immich_server):
    config, sf, lib = _env(
        tmp_path,
        immich_server,
        uploader=_uploader(tmp_path, "i=0; while [ $i -lt 2000 ]; do echo line-$i; i=$((i+1)); done"),
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)

    from photo_pipeline.integrations import immich_go

    service = _service(sf, config)
    original = immich_go.MAX_REPORT_BYTES
    immich_go.MAX_REPORT_BYTES = 500  # a cap small enough to hit in one test
    try:
        result = service.run(batch["id"])
    finally:
        immich_go.MAX_REPORT_BYTES = original

    assert result["state"] == BatchState.SUCCEEDED, "the uploader still finished normally"
    assert result["report_truncated"] is True
    assert result["report_bytes"] <= 500
    assert Path(result["report_path"]).stat().st_size <= 500


# ── one lane ─────────────────────────────────────────────────────────────────


def test_a_second_batch_cannot_run_while_one_is_running(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))
    rome, paris = sorted(_approved(sf, config), key=lambda b: b["album"] != "rome")
    with sf() as session:  # a batch already occupying the lane
        session.get(UploadBatch, rome["id"]).state = BatchState.RUNNING
        session.commit()

    with pytest.raises(BatchConflict) as error:
        _service(sf, config).run(paris["id"])

    assert error.value.code == "lane_busy"
    assert _argv(tmp_path) == []


def test_the_job_lock_refuses_a_second_queued_upload(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib, "rome")
    _album(sf, lib, "paris", names=("c.jpg",))
    batches = _approved(sf, config)

    with TestClient(create_app(config)) as client:
        first = client.post(f"/api/v1/upload-batches/{batches[0]['id']}/start")
        second = client.post(f"/api/v1/upload-batches/{batches[1]['id']}/start")

    assert first.status_code == 200
    assert second.status_code == 409 and second.json()["error"]["code"] == "lock_held"


# ── cancellation ─────────────────────────────────────────────────────────────


def test_cancelling_a_running_batch_stops_the_uploader(tmp_path, immich_server):
    """The uploader sleeps; cancelling flips the batch to ``cancelling`` and the
    running attempt must terminate the process and record ``cancelled``."""
    config, sf, lib = _env(
        tmp_path, immich_server, uploader=_uploader(tmp_path, 'echo started; sleep 30; exit 0')
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    outcome = {}

    def _run():
        outcome["batch"] = service.run(batch["id"])

    runner = threading.Thread(target=_run)
    started = time.monotonic()
    runner.start()
    while service.get(batch["id"])["state"] != BatchState.RUNNING:
        assert time.monotonic() - started < 30, "the attempt never started"
        time.sleep(0.02)

    service.cancel(batch["id"])
    runner.join(timeout=30)

    assert not runner.is_alive(), "cancellation must not wait for the uploader's own timeout"
    assert outcome["batch"]["state"] == BatchState.CANCELLED
    assert all(item["state"] == ItemState.PENDING for item in outcome["batch"]["items"])


def test_cancelling_a_planned_batch_never_starts_the_uploader(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)

    cancelled = service.cancel(batch["id"])

    assert cancelled["state"] == BatchState.CANCELLED
    assert _argv(tmp_path) == []


def test_a_cancelled_batch_can_be_run_again(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    service.cancel(batch["id"])

    result = service.run(batch["id"])

    assert result["state"] == BatchState.SUCCEEDED and result["attempt_count"] == 1


# ── restart / recovery ───────────────────────────────────────────────────────


def test_an_interrupted_attempt_becomes_uncertain_not_failed(tmp_path, immich_server):
    """The process died mid-upload: Immich may hold the files, so the batch requires
    verification (US05-04) instead of a blind retry, and the lane is released."""
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    with sf() as session:  # what a killed worker leaves behind
        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
        session.commit()
    service = _service(sf, config)

    assert service.recover() == {"interrupted": 1}

    recovered = service.get(batch["id"])
    assert recovered["state"] == BatchState.UNKNOWN
    assert recovered["error_code"] == "interrupted"
    with pytest.raises(BatchError) as error:
        service.run(batch["id"])
    assert error.value.code == "requires_verification"
    assert _argv(tmp_path) == []


def test_recovery_leaves_a_planned_batch_runnable(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)

    assert service.recover() == {"interrupted": 0}

    assert service.run(batch["id"])["state"] == BatchState.SUCCEEDED


def test_application_startup_recovers_an_interrupted_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    with sf() as session:
        session.get(UploadBatch, batch["id"]).state = BatchState.RUNNING
        session.commit()

    with TestClient(create_app(config)) as client:
        state = client.get(f"/api/v1/upload-batches/{batch['id']}").json()["state"]

    assert state == BatchState.UNKNOWN


def test_a_killed_uploader_leaves_an_uncertain_batch(tmp_path, immich_server):
    """A real SIGKILL of the uploader process, not a simulated state write."""
    config, sf, lib = _env(
        tmp_path,
        immich_server,
        uploader=_uploader(tmp_path, 'echo "pid $$"; sleep 30; exit 0'),
    )
    _album(sf, lib)
    (batch,) = _approved(sf, config)
    service = _service(sf, config)
    outcome = {}
    runner = threading.Thread(target=lambda: outcome.update(batch=service.run(batch["id"])))
    runner.start()

    report = tmp_path / "data" / "uploads"
    deadline = time.monotonic() + 30
    pid = None
    while pid is None:
        assert time.monotonic() < deadline, "the uploader never announced itself"
        for log in report.glob("*.log"):
            text = log.read_text()
            if text.startswith("pid "):
                pid = int(text.split()[1])
        time.sleep(0.02)
    os.kill(pid, signal.SIGKILL)
    runner.join(timeout=30)

    # The process is gone with a non-zero status and no parsed report: the attempt
    # failed locally, and a restart classifies it honestly.
    assert outcome["batch"]["state"] in (BatchState.FAILED, BatchState.UNKNOWN)
    assert outcome["batch"]["exit_code"] != 0


# ── worker integration ───────────────────────────────────────────────────────


def test_the_worker_runs_a_queued_batch_on_the_upload_lane(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)
    (batch,) = _approved(sf, config)

    with TestClient(create_app(config)) as client:
        response = client.post(f"/api/v1/upload-batches/{batch['id']}/start")
    job_id = response.json()["job"]["id"]
    Worker(sf, worker_id="uploader-1", job_types=[UPLOAD_BATCH], config=config).run_once()

    assert JobService(sf).get(job_id)["state"] == JobState.SUCCEEDED
    assert _service(sf, config).get(batch["id"])["state"] == BatchState.SUCCEEDED
    assert len(_argv(tmp_path)) == 1


# ── API surface ──────────────────────────────────────────────────────────────


def test_api_creates_lists_and_reads_batches_without_secrets(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)

    with TestClient(create_app(config)) as client:
        token = client.post("/api/v1/upload-preflight", json={}).json()["token"]
        created = client.post("/api/v1/upload-batches", json={"token": token})
        listed = client.get("/api/v1/upload-batches")
        batch_id = created.json()["batches"][0]["id"]
        fetched = client.get(f"/api/v1/upload-batches/{batch_id}")

    assert created.status_code == 201
    assert listed.json()["batches"][0]["id"] == batch_id
    assert fetched.json()["state"] == BatchState.PLANNED
    assert SENTINEL_KEY not in created.text + listed.text + fetched.text


def test_api_rejects_a_stale_token_and_an_unknown_batch(tmp_path, immich_server):
    config, sf, lib = _env(tmp_path, immich_server)
    _album(sf, lib)

    with TestClient(create_app(config)) as client:
        stale = client.post("/api/v1/upload-batches", json={"token": "v1:not-the-token"})
        missing = client.get("/api/v1/upload-batches/does-not-exist")

    assert stale.status_code == 409 and stale.json()["error"]["code"] == "stale_preflight"
    assert missing.status_code == 404