US02-04: Expose Job APIs and Activity Events

tests/integration/test_jobs_api.py

@@ -0,0 +1,108 @@
+"""Job API contract: typed commands, status codes, error envelope, idempotency,
+cancellation, blockers, and OpenAPI schema."""
+
+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 run_migrations
+from photo_pipeline.services.jobs import ItemState, JobService, JobState
+
+
+@pytest.fixture
+def client(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    app = create_app(config)
+    with TestClient(app) as test_client:
+        test_client.app = app
+        yield test_client
+
+
+def _drive_to_success(app, job_id, items):
+    service = JobService(app.state.session_factory)
+    service.claim(["scan"], "w1")
+    for item in items:
+        service.set_item(job_id, item, ItemState.RUNNING)
+        service.set_item(job_id, item, ItemState.SUCCEEDED)
+    service.transition(job_id, JobState.SUCCEEDED, worker_id="w1")
+
+
+def test_start_status_and_progress(client):
+    start = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]})
+    assert start.status_code == 200
+    job = start.json()
+    assert job["state"] == "queued"
+
+    status = client.get(f"/api/v1/jobs/{job['id']}")
+    assert status.status_code == 200
+    body = status.json()
+    assert body["progress"] == {"total": 2, "done": 0, "by_state": {"queued": 2}}
+
+
+def test_idempotent_start_returns_same_job(client):
+    first = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json()
+    second = client.post("/api/v1/jobs", json={"job_type": "scan", "idempotency_key": "k"}).json()
+    assert first["id"] == second["id"]
+
+
+def test_lock_conflict_returns_409(client):
+    client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"})
+    conflict = client.post("/api/v1/jobs", json={"job_type": "rename", "lock": "library_write"})
+    assert conflict.status_code == 409
+    assert conflict.json()["error"]["code"] == "lock_held"
+
+
+def test_cancel_queued_then_terminal(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    cancelled = client.post(f"/api/v1/jobs/{job['id']}/cancel")
+    assert cancelled.status_code == 200
+    assert cancelled.json()["state"] == "cancelled"
+    # Cancelling a terminal job conflicts.
+    again = client.post(f"/api/v1/jobs/{job['id']}/cancel")
+    assert again.status_code == 409
+
+
+def test_blockers_endpoint(client):
+    client.post("/api/v1/jobs", json={"job_type": "scan", "lock": "library_write"})
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    blockers = client.get(f"/api/v1/jobs/{job['id']}/blockers?lock=library_write").json()
+    assert len(blockers["blockers"]) == 1
+
+
+def test_unknown_job_uses_error_envelope(client):
+    for resp in (
+        client.get("/api/v1/jobs/ghost"),
+        client.post("/api/v1/jobs/ghost/cancel"),
+        client.get("/api/v1/jobs/ghost/events"),
+    ):
+        assert resp.status_code == 404
+        assert resp.json()["error"]["code"] == "not_found"
+
+
+def test_polling_events_with_cursor(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a"]}).json()
+    _drive_to_success(client.app, job["id"], ["a"])
+    all_events = client.get(f"/api/v1/jobs/{job['id']}/events").json()["events"]
+    seqs = [e["seq"] for e in all_events]
+    assert seqs == sorted(seqs) and len(seqs) >= 3
+    after_first = client.get(f"/api/v1/jobs/{job['id']}/events?after={seqs[0]}").json()["events"]
+    assert [e["seq"] for e in after_first] == seqs[1:]
+
+
+def test_openapi_exposes_job_contracts(client):
+    paths = client.get("/openapi.json").json()["paths"]
+    assert "/api/v1/jobs" in paths
+    assert "post" in paths["/api/v1/jobs"]
+    assert "/api/v1/jobs/{job_id}" in paths
+    assert "/api/v1/jobs/{job_id}/cancel" in paths
+    assert "/api/v1/jobs/{job_id}/events/stream" in paths
+
+
+def test_status_content_type_is_json(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan"}).json()
+    assert client.get(f"/api/v1/jobs/{job['id']}").headers["content-type"].startswith(
+        "application/json"
+    )

tests/integration/test_jobs_sse.py

@@ -0,0 +1,69 @@
+"""SSE activity stream: resumable event ids, reconnect without duplicate/lost
+events, and parity with the polling fallback."""
+
+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 run_migrations
+from photo_pipeline.services.jobs import ItemState, JobService, JobState
+
+
+@pytest.fixture
+def client(tmp_path):
+    (tmp_path / "data").mkdir()
+    config = Config.from_env({"PHOTO_PIPELINE_DATA_DIR": str(tmp_path / "data")})
+    run_migrations(config.database_url)
+    app = create_app(config)
+    with TestClient(app) as test_client:
+        test_client.app = app
+        yield test_client
+
+
+def _finished_job(client):
+    job = client.post("/api/v1/jobs", json={"job_type": "scan", "items": ["a", "b"]}).json()
+    service = JobService(client.app.state.session_factory)
+    service.claim(["scan"], "w1")
+    for item in ("a", "b"):
+        service.set_item(job["id"], item, ItemState.RUNNING)
+        service.set_item(job["id"], item, ItemState.SUCCEEDED)
+    service.transition(job["id"], JobState.SUCCEEDED, worker_id="w1")
+    return job["id"]
+
+
+def _read_ids(client, job_id, headers=None):
+    ids, saw_done = [], False
+    with client.stream(
+        "GET", f"/api/v1/jobs/{job_id}/events/stream", headers=headers or {}
+    ) as stream:
+        assert stream.headers["content-type"].startswith("text/event-stream")
+        for line in stream.iter_lines():
+            if line.startswith("id:"):
+                ids.append(int(line.split(":", 1)[1]))
+            elif line.startswith("event: done"):
+                saw_done = True
+    return ids, saw_done
+
+
+def test_sse_streams_all_events_then_closes(client):
+    job_id = _finished_job(client)
+    ids, saw_done = _read_ids(client, job_id)
+    assert ids == sorted(ids) and len(ids) >= 3
+    assert saw_done  # terminal job closes the stream
+
+
+def test_sse_reconnect_has_no_duplicate_or_lost_events(client):
+    job_id = _finished_job(client)
+    full, _ = _read_ids(client, job_id)
+    resume_from = full[1]
+    resumed, saw_done = _read_ids(client, job_id, headers={"Last-Event-ID": str(resume_from)})
+    assert resumed == [i for i in full if i > resume_from]  # no dup, none lost
+    assert saw_done
+
+
+def test_sse_matches_polling(client):
+    job_id = _finished_job(client)
+    sse_ids, _ = _read_ids(client, job_id)
+    polled = client.get(f"/api/v1/jobs/{job_id}/events").json()["events"]
+    assert sse_ids == [e["seq"] for e in polled]