Add audit-emission test coverage for External Results write endpoints + spec doc update

backend/tests/api/test_external_results_audit.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import io
 import uuid
 from types import SimpleNamespace
 
@@ -10,6 +11,7 @@
 from fastapi.testclient import TestClient
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 
+from app.api.external_results import router as external_results_router
 from app.auth.dependencies import get_current_user
 from app.crud.audit_log import create_audit_entry
 from app.crud.runner_token import create_runner_token
@@ -18,6 +20,7 @@
 from app.models.base import Base
 from app.models.project import Project
 from app.models.user import User, UserRole
+from app.storage.base import StorageResult
 
 _PROJECT_ID = str(uuid.uuid4())
 
@@ -209,3 +212,356 @@ async def override_admin():
     data = response.json()
     assert data["total"] >= 1
     assert any(entry["actor_kind"] == "user" for entry in data["entries"])
+
+
+# ---------------------------------------------------------------------------
+# Per-endpoint audit-emission tests
+# ---------------------------------------------------------------------------
+
+
+def test_case_create_writes_audit(monkeypatch, db_session, write_token):
+    captured: list[dict] = []
+
+    async def fake_write_audit(db, **kwargs):
+        captured.append(kwargs)
+        return SimpleNamespace(id=uuid.uuid4())
+
+    monkeypatch.setattr("app.api.external_results.write_audit", fake_write_audit)
+
+    token_model, plaintext = write_token
+    with TestClient(app) as client:
+        session_resp = client.post(
+            "/api/v1/external-results/session",
+            json=_session_payload(),
+            headers=_auth_header(plaintext),
+        )
+        assert session_resp.status_code == 201, session_resp.text
+        session_id = session_resp.json()["id"]
+
+        resp = client.post(
+            "/api/v1/external-results/case",
+            json={
+                "session_id": session_id,
+                "external_id": f"TC-{uuid.uuid4()}",
+                "title": "Audit test case",
+                "outcome": "passed",
+                "duration_ms": 50,
+                "requirement_ids": [],
+                "requirement_external_ids": ["REQ-001"],
+            },
+            headers=_auth_header(plaintext),
+        )
+
+    assert resp.status_code == 201, resp.text
+    create_audit = next((c for c in captured if c["action"] == "external_results.case.create"), None)
+    assert create_audit is not None
+    assert create_audit["actor_kind"] == "runner_token"
+    assert create_audit["actor_id"] == token_model.id
+    details = create_audit["details"]
+    assert "session_id" in details
+    assert "outcome" in details
+    assert "external_id" in details
+    assert "auto_registered" in details
+
+
+def test_case_create_idempotent_writes_separate_audit(monkeypatch, db_session, write_token):
+    captured: list[dict] = []
+
+    async def fake_write_audit(db, **kwargs):
+        captured.append(kwargs)
+        return SimpleNamespace(id=uuid.uuid4())
+
+    monkeypatch.setattr("app.api.external_results.write_audit", fake_write_audit)
+
+    _token_model, plaintext = write_token
+    with TestClient(app) as client:
+        session_resp = client.post(
+            "/api/v1/external-results/session",
+            json=_session_payload(),
+            headers=_auth_header(plaintext),
+        )
+        assert session_resp.status_code == 201, session_resp.text
+        session_id = session_resp.json()["id"]
+
+        case_payload = {
+            "session_id": session_id,
+            "external_id": f"TC-idem-{uuid.uuid4()}",
+            "title": "Idempotent test case",
+            "outcome": "passed",
+            "duration_ms": 30,
+            "requirement_ids": [],
+        }
+        first = client.post("/api/v1/external-results/case", json=case_payload, headers=_auth_header(plaintext))
+        second = client.post("/api/v1/external-results/case", json=case_payload, headers=_auth_header(plaintext))
+
+    assert first.status_code == 201, first.text
+    assert second.status_code == 200, second.text
+    create_audits = [c for c in captured if c["action"] == "external_results.case.create"]
+    idempotent_audits = [c for c in captured if c["action"] == "external_results.case.create.idempotent"]
+    assert len(create_audits) == 1
+    assert len(idempotent_audits) == 1
+
+
+def test_case_update_writes_audit(monkeypatch, db_session, write_token):
+    captured: list[dict] = []
+
+    async def fake_write_audit(db, **kwargs):
+        captured.append(kwargs)
+        return SimpleNamespace(id=uuid.uuid4())
+
+    monkeypatch.setattr("app.api.external_results.write_audit", fake_write_audit)
+
+    _token_model, plaintext = write_token
+    with TestClient(app) as client:
+        session_resp = client.post(
+            "/api/v1/external-results/session",
+            json=_session_payload(),
+            headers=_auth_header(plaintext),
+        )
+        assert session_resp.status_code == 201, session_resp.text
+        session_id = session_resp.json()["id"]
+
+        create_resp = client.post(
+            "/api/v1/external-results/case",
+            json={
+                "session_id": session_id,
+                "external_id": f"TC-upd-{uuid.uuid4()}",
+                "title": "Update test case",
+                "outcome": "passed",
+                "duration_ms": 30,
+                "requirement_ids": [],
+            },
+            headers=_auth_header(plaintext),
+        )
+        assert create_resp.status_code == 201, create_resp.text
+        case_id = create_resp.json()["id"]
+
+        patch_resp = client.patch(
+            f"/api/v1/external-results/case/{case_id}",
+            json={"outcome": "flaky"},
+            headers=_auth_header(plaintext),
+        )
+
+    assert patch_resp.status_code == 200, patch_resp.text
+    update_audit = next((c for c in captured if c["action"] == "external_results.case.update"), None)
+    assert update_audit is not None
+    details = update_audit["details"]
+    assert "previous_outcome" in details
+    assert "new_outcome" in details
+    assert details["previous_outcome"] == "passed"
+    assert details["new_outcome"] == "flaky"
+
+
+def test_artifact_upload_writes_audit(monkeypatch, db_session, write_token):
+    captured: list[dict] = []
+
+    async def fake_write_audit(db, **kwargs):
+        captured.append(kwargs)
+        return SimpleNamespace(id=uuid.uuid4())
+
+    monkeypatch.setattr("app.api.external_results.write_audit", fake_write_audit)
+
+    fake_storage_result = StorageResult(
+        key="test/artifact/test.png",
+        url="http://example.com/test.png",
+        size_bytes=72,
+        content_type="image/png",
+    )
+    monkeypatch.setattr(
+        "app.api.external_results.get_storage",
+        lambda: SimpleNamespace(save=lambda stream, *, key, content_type: fake_storage_result),
+    )
+
+    _token_model, plaintext = write_token
+    with TestClient(app) as client:
+        session_resp = client.post(
+            "/api/v1/external-results/session",
+            json=_session_payload(),
+            headers=_auth_header(plaintext),
+        )
+        assert session_resp.status_code == 201, session_resp.text
+        session_id = session_resp.json()["id"]
+
+        create_resp = client.post(
+            "/api/v1/external-results/case",
+            json={
+                "session_id": session_id,
+                "external_id": f"TC-art-{uuid.uuid4()}",
+                "title": "Artifact test case",
+                "outcome": "passed",
+                "duration_ms": 30,
+                "requirement_ids": [],
+            },
+            headers=_auth_header(plaintext),
+        )
+        assert create_resp.status_code == 201, create_resp.text
+        case_id = create_resp.json()["id"]
+
+        png_bytes = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
+        upload_resp = client.post(
+            "/api/v1/external-results/artifact",
+            data={"case_result_id": case_id, "kind": "screenshot", "filename": "test.png"},
+            files={"file": ("test.png", io.BytesIO(png_bytes), "image/png")},
+            headers=_auth_header(plaintext),
+        )
+
+    assert upload_resp.status_code == 201, upload_resp.text
+    artifact_audit = next((c for c in captured if c["action"] == "external_results.artifact.upload"), None)
+    assert artifact_audit is not None
+    details = artifact_audit["details"]
+    assert "case_result_id" in details
+    assert "kind" in details
+    assert "filename" in details
+    assert "content_type" in details
+    assert "size_bytes" in details
+
+
+# ---------------------------------------------------------------------------
+# Regression-proof enforcement test
+# ---------------------------------------------------------------------------
+
+# Endpoints that intentionally do NOT audit (reads):
+_READ_ONLY_OPS = frozenset(
+    {
+        ("GET", "/external-results/session/{session_id}"),
+        ("GET", "/external-results/case/{case_result_id}"),
+    }
+)
+
+
+def _state_changing_routes():
+    """Yield (method, path, route) for every non-GET endpoint in the router."""
+    for route in external_results_router.routes:
+        if not hasattr(route, "methods"):
+            continue
+        for method in route.methods:
+            if method == "HEAD":
+                continue
+            if (method, route.path) in _READ_ONLY_OPS:
+                continue
+            if method == "GET":
+                continue
+            yield method, route.path, route
+
+
+@pytest.mark.parametrize("method,path,_route", list(_state_changing_routes()))
+def test_state_changing_endpoint_emits_audit(monkeypatch, db_session, write_token, method, path, _route):
+    """Every state-changing endpoint MUST call write_audit at least once.
+
+    If a new endpoint is added without an audit call, this test fails by default
+    — that's the point. To intentionally skip auditing on a new read endpoint,
+    add it to _READ_ONLY_OPS.
+    """
+    captured: list[dict] = []
+
+    async def fake_write_audit(db, **kwargs):
+        captured.append(kwargs)
+        return SimpleNamespace(id=uuid.uuid4())
+
+    monkeypatch.setattr("app.api.external_results.write_audit", fake_write_audit)
+
+    # Mock storage for the artifact upload endpoint.
+    fake_storage_result = StorageResult(
+        key="enf/test.png",
+        url="http://example.com/enf/test.png",
+        size_bytes=72,
+        content_type="image/png",
+    )
+    monkeypatch.setattr(
+        "app.api.external_results.get_storage",
+        lambda: SimpleNamespace(save=lambda stream, *, key, content_type: fake_storage_result),
+    )
+
+    token_model, plaintext = write_token
+    headers = _auth_header(plaintext)
+
+    with TestClient(app) as client:
+
+        def _create_session_id() -> str:
+            r = client.post("/api/v1/external-results/session", json=_session_payload(), headers=headers)
+            assert r.status_code == 201, r.text
+            return r.json()["id"]
+
+        def _create_case_id(session_id: str) -> str:
+            r = client.post(
+                "/api/v1/external-results/case",
+                json={
+                    "session_id": session_id,
+                    "external_id": f"enf-{uuid.uuid4()}",
+                    "title": "enforcement test case",
+                    "outcome": "passed",
+                    "duration_ms": 5,
+                    "requirement_ids": [],
+                },
+                headers=headers,
+            )
+            assert r.status_code == 201, r.text
+            return r.json()["id"]
+
+        # Drive the endpoint with a happy-path call.
+        # captured.clear() is called immediately before each actual endpoint call so that
+        # prerequisite setup calls (which also invoke fake_write_audit) do not mask a
+        # missing write_audit on the endpoint under test.
+        if (method, path) == ("POST", "/external-results/session"):
+            captured.clear()
+            resp = client.post("/api/v1/external-results/session", json=_session_payload(), headers=headers)
+
+        elif (method, path) == ("PATCH", "/external-results/session/{session_id}"):
+            session_id = _create_session_id()
+            captured.clear()
+            resp = client.patch(
+                f"/api/v1/external-results/session/{session_id}",
+                json={"status": "passed", "summary": {"total": 1, "passed": 1}},
+                headers=headers,
+            )
+
+        elif (method, path) == ("POST", "/external-results/case"):
+            session_id = _create_session_id()
+            captured.clear()
+            resp = client.post(
+                "/api/v1/external-results/case",
+                json={
+                    "session_id": session_id,
+                    "external_id": f"enf-{uuid.uuid4()}",
+                    "title": "enforcement test case",
+                    "outcome": "passed",
+                    "duration_ms": 5,
+                    "requirement_ids": [],
+                },
+                headers=headers,
+            )
+
+        elif (method, path) == ("PATCH", "/external-results/case/{case_result_id}"):
+            session_id = _create_session_id()
+            case_id = _create_case_id(session_id)
+            captured.clear()
+            resp = client.patch(
+                f"/api/v1/external-results/case/{case_id}",
+                json={"outcome": "flaky"},
+                headers=headers,
+            )
+
+        elif (method, path) == ("POST", "/external-results/artifact"):
+            session_id = _create_session_id()
+            case_id = _create_case_id(session_id)
+            captured.clear()
+            png_bytes = b"\x89PNG\r\n\x1a\n" + b"\x00" * 64
+            resp = client.post(
+                "/api/v1/external-results/artifact",
+                data={"case_result_id": case_id, "kind": "screenshot", "filename": "test.png"},
+                files={"file": ("test.png", io.BytesIO(png_bytes), "image/png")},
+                headers=headers,
+            )
+
+        else:
+            raise AssertionError(
+                f"No dispatch driver for {method} {path}. "
+                "Add a happy-path driver to test_state_changing_endpoint_emits_audit."
+            )
+
+    assert resp.status_code in (200, 201), f"{method} {path} returned {resp.status_code}: {resp.text}"
+    assert len(captured) >= 1, (
+        f"{method} {path} did not call write_audit. "
+        f"Every state-changing External Results endpoint must emit an audit entry. "
+        f"If this endpoint is intentionally read-only, add it to _READ_ONLY_OPS."
+    )

docs/specs/external_results_v1.md

@@ -566,6 +566,7 @@ Every state-changing call now writes an **audit log entry** recording actor iden
 | `external_results.session.start` | `POST /session` → `201` |
 | `external_results.session.finish` | `PATCH /session/{id}` → `200` |
 | `external_results.case.create` | `POST /case` → `201` |
+| `external_results.case.create.idempotent` | `POST /case` returning `200` (duplicate `external_id`) |
 | `external_results.case.update` | `PATCH /case/{id}` → `200` |
 | `external_results.artifact.upload` | `POST /artifact` → `201` |