"""HTTP assertions — MUSEHUB-F7b overseer-run provenance (freeze 2026-08-11). Four TestClient cases across the real app route table. Registration is not modified. Postgres is not required: tests/unit/ no-ops the session DB fixture and this module overrides only get_db (optional_token pulls it transitively). """ from __future__ import annotations from collections.abc import AsyncIterator, Iterator import pytest from fastapi.testclient import TestClient from musehub.db.database import get_db from musehub.main import app from musehub.services import musehub_overseer_provenance as enrichment FIXTURE_REF = "flow_run:fixture-overseer-001" UNKNOWN_REF = "flow_run:not-seeded-0001" PREFIXED_PATH = f"/api/overseer-run-provenance/{FIXTURE_REF}" ENCODED_PATH = "/api/overseer-run-provenance/flow_run%3Afixture-overseer-001" EXPECTED_RECORD: dict[str, object] = { "schema": "scooling.overseer_run_provenance/v0", "contractVersion": "scooling.overseer_provenance_enrichment/v0", "runRef": FIXTURE_REF, "reviewTrayOutcomeRef": f"outcome:{FIXTURE_REF}", "outcome": "pass", "constitutionVersion": "0.1.0", "agentVersionRef": "agent:fixture-overseer-v0.1.0", "workerModelFamily": "family:ollama", "checkerModelFamily": "family:llama_cpp", "externalRef": None, "actorHash": "a" * 64, "producedAt": "2026-07-09T12:00:00Z", "untrusted": True, } RECORD_KEYS = frozenset(EXPECTED_RECORD.keys()) TOP_LEVEL_KEYS = frozenset({"schema", "record"}) @pytest.fixture def client() -> Iterator[TestClient]: """Plain TestClient (no lifespan) with get_db overridden only.""" async def _override_get_db() -> AsyncIterator[object]: yield object() app.dependency_overrides[get_db] = _override_get_db try: yield TestClient(app) finally: app.dependency_overrides.pop(get_db, None) def _enable_posture(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( enrichment.settings, "musehub_overseer_provenance_enrichment_enabled", "enabled", ) def _disable_posture(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr( enrichment.settings, "musehub_overseer_provenance_enrichment_enabled", "disabled", ) def test_a1_prefixed_path_returns_envelope( client: TestClient, monkeypatch: pytest.MonkeyPatch ) -> None: """A1 — posture enabled; /api/…/flow_run:fixture → 200 with exact envelope.""" _enable_posture(monkeypatch) response = client.get(PREFIXED_PATH) assert response.status_code == 200 body = response.json() assert set(body.keys()) == TOP_LEVEL_KEYS assert body["schema"] == "musehub.overseer_run_provenance_envelope/v0" record = body["record"] assert set(record.keys()) == RECORD_KEYS assert record == EXPECTED_RECORD def test_a3_unknown_ref_returns_404( client: TestClient, monkeypatch: pytest.MonkeyPatch ) -> None: """A3 — posture enabled; well-formed unknown ref → 404 (store miss).""" _enable_posture(monkeypatch) response = client.get(f"/api/overseer-run-provenance/{UNKNOWN_REF}") assert response.status_code == 404 def test_a2_percent_encoded_ref_returns_decoded_record( client: TestClient, monkeypatch: pytest.MonkeyPatch ) -> None: """A2 — percent-encoded ref → 200; runRef decoded; full record equals A1.""" _enable_posture(monkeypatch) response = client.get(ENCODED_PATH) assert response.status_code == 200 body = response.json() assert set(body.keys()) == TOP_LEVEL_KEYS assert body["schema"] == "musehub.overseer_run_provenance_envelope/v0" record = body["record"] assert record["runRef"] == FIXTURE_REF assert record == EXPECTED_RECORD def test_a4_posture_off_returns_404_fail_closed( client: TestClient, monkeypatch: pytest.MonkeyPatch ) -> None: """A4 — posture disabled + known-good ref → 404; no record field leakage.""" _disable_posture(monkeypatch) response = client.get(PREFIXED_PATH) assert response.status_code == 404 body_text = response.text assert "runRef" not in body_text assert "outcome" not in body_text assert "actorHash" not in body_text assert FIXTURE_REF not in body_text