cyclone/backend/tests/test_api_audit_log.py

"""Tests for the audit-log admin API endpoints. SP11 T3."""
from __future__ import annotations

import json

import pytest
from fastapi.testclient import TestClient

from cyclone import db
from cyclone.audit_log import AuditEvent, append_event
from cyclone.api import app
from cyclone.db import AuditLog, init_db


@pytest.fixture(autouse=True)
def _fresh_db():
    init_db()
    yield


@pytest.fixture
def client() -> TestClient:
    return TestClient(app)


def _seed_audit(rows: list[AuditEvent]) -> None:
    with db.SessionLocal()() as s:
        for ev in rows:
            append_event(s, ev)
        s.commit()


# --------------------------------------------------------------------------- #
# List endpoint
# --------------------------------------------------------------------------- #


class TestListAuditLog:
    def test_empty_returns_zero_items(self, client: TestClient):
        r = client.get("/api/admin/audit-log")
        assert r.status_code == 200
        body = r.json()
        assert body["total"] == 0
        assert body["items"] == []

    def test_returns_newest_first(self, client: TestClient):
        _seed_audit([
            AuditEvent(event_type="a.first", entity_type="claim", entity_id="c1"),
            AuditEvent(event_type="a.second", entity_type="claim", entity_id="c1"),
            AuditEvent(event_type="a.third", entity_type="claim", entity_id="c1"),
        ])
        r = client.get("/api/admin/audit-log")
        assert r.status_code == 200
        items = r.json()["items"]
        assert [i["event_type"] for i in items] == ["a.third", "a.second", "a.first"]

    def test_filter_by_entity(self, client: TestClient):
        _seed_audit([
            AuditEvent(event_type="x", entity_type="claim", entity_id="c1"),
            AuditEvent(event_type="x", entity_type="claim", entity_id="c2"),
            AuditEvent(event_type="x", entity_type="batch", entity_id="b1"),
        ])
        r = client.get("/api/admin/audit-log", params={"entity_type": "claim", "entity_id": "c1"})
        items = r.json()["items"]
        assert len(items) == 1
        assert items[0]["entity_id"] == "c1"

    def test_filter_by_event_type(self, client: TestClient):
        _seed_audit([
            AuditEvent(event_type="claim.parsed", entity_type="claim", entity_id="c1"),
            AuditEvent(event_type="claim.rejected", entity_type="claim", entity_id="c1"),
        ])
        r = client.get("/api/admin/audit-log", params={"event_type": "claim.rejected"})
        items = r.json()["items"]
        assert len(items) == 1
        assert items[0]["event_type"] == "claim.rejected"


# --------------------------------------------------------------------------- #
# Verify endpoint
# --------------------------------------------------------------------------- #


class TestVerifyAuditLog:
    def test_empty_chain_returns_ok(self, client: TestClient):
        r = client.get("/api/admin/audit-log/verify")
        assert r.status_code == 200
        body = r.json()
        assert body["ok"] is True
        assert body["checked"] == 0

    def test_clean_chain_returns_ok(self, client: TestClient):
        _seed_audit([
            AuditEvent(event_type="x", entity_type="y", entity_id=f"id-{i}")
            for i in range(10)
        ])
        r = client.get("/api/admin/audit-log/verify")
        body = r.json()
        assert body["ok"] is True
        assert body["checked"] == 10

    def test_tampered_chain_returns_failure(self, client: TestClient):
        _seed_audit([
            AuditEvent(event_type="x", entity_type="y", entity_id=f"id-{i}")
            for i in range(5)
        ])
        with db.SessionLocal()() as s:
            row = s.query(AuditLog).filter(AuditLog.id == 3).first()
            row.payload_json = json.dumps({"evil": True})
            s.commit()
        r = client.get("/api/admin/audit-log/verify")
        body = r.json()
        assert body["ok"] is False
        assert body["first_bad_id"] == 3
        assert body["reason"]


# --------------------------------------------------------------------------- #
# End-to-end: 999 → audit row appears in list
# --------------------------------------------------------------------------- #


class TestAuditLogHookedIntoEndpoints:
    def test_parse_999_creates_audit_rows(self, client: TestClient):
        """Upload a 999 with one rejected set, see an audit row appear."""
        # Seed a claim so the 999 rejection matches.
        from cyclone.db import Claim, ClaimState
        with db.SessionLocal()() as s:
            s.add(Claim(
                id="c1", batch_id="B-1", patient_control_number="0001",
                state=ClaimState.SUBMITTED,
            ))
            s.commit()

        text = (
            "ISA*00*          *00*          *ZZ*AAAAAAAAAAAAAAA*ZZ*BBBBBBBBBBBBBBB*240620*1200*^*00501*000000001*0*P*:~"
            "GS*HN*A*B*20240620*1200*1*X*005010X231A1~"
            "ST*999*0001*005010X231A1~"
            "AK1*HC*0001~"
            "AK2*837*0001~"
            "AK5*R~"
            "AK9*R*1*0*1~"
            "SE*7*0001~"
            "GE*1*1~"
            "IEA*1*000000001~"
        )
        r = client.post(
            "/api/parse-999",
            files={"file": ("r.txt", text, "text/plain")},
        )
        assert r.status_code == 200, r.text

        # One audit row should have appeared for the rejected claim.
        r2 = client.get(
            "/api/admin/audit-log",
            params={"entity_type": "claim", "entity_id": "c1"},
        )
        items = r2.json()["items"]
        assert any(i["event_type"] == "claim.rejected" for i in items)