diff --git a/backend/tests/api/test_external_results_audit.py b/backend/tests/api/test_external_results_audit.py index f49381d..e6d12b6 100644 --- a/backend/tests/api/test_external_results_audit.py +++ b/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." + ) diff --git a/docs/specs/external_results_v1.md b/docs/specs/external_results_v1.md index f0c1669..66fd193 100644 --- a/docs/specs/external_results_v1.md +++ b/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` |