gendesign/backend/tests/test_audit_middleware.py

"""§19 audit-log middleware tests (#962, EPIC18).

Coverage:
  - classify_path: analyze / forecast / forecast-export матчатся, cad_num
    извлекается; forecast/export НЕ путается с forecast; non-matched → None.
  - write_audit_row: best-effort INSERT — ровно одна строка с верными полями;
    сбой записи (execute кидает) проглатывается, не re-raise; rollback вызван.
  - middleware (копия, минующая settings.testing-bypass, как в test_rbac.py):
    * matched request → ровно один audit write с верными полями + реальный
      status_code из ответа.
    * non-matched path → НИ одной записи.
    * сбой audit write → запрос всё равно отдаёт настоящий response.

DB не нужен: SessionLocal патчится фейковой сессией, ловящей execute()/commit().
auth.get_role патчится, чтобы не зависеть от roles.yaml-кэша.
"""

from __future__ import annotations

from collections.abc import Awaitable, Callable
from typing import Any

import pytest
from fastapi import FastAPI, Request
from fastapi.responses import Response
from fastapi.testclient import TestClient

from app.core import audit_middleware as audit_mod

# ---------------------------------------------------------------------------
# Fake DB session — записывает execute()/commit()/rollback()/close().
# ---------------------------------------------------------------------------


class _FakeSession:
    def __init__(self, *, fail_on_execute: bool = False) -> None:
        self.executed: list[tuple[str, dict[str, Any]]] = []
        self.committed = 0
        self.rolled_back = 0
        self.closed = 0
        self._fail_on_execute = fail_on_execute

    def execute(self, stmt: Any, params: dict[str, Any]) -> None:
        if self._fail_on_execute:
            raise RuntimeError("simulated DB failure")
        # stmt — SQLAlchemy TextClause; .text несёт сырой SQL.
        self.executed.append((str(getattr(stmt, "text", stmt)), params))

    def commit(self) -> None:
        self.committed += 1

    def rollback(self) -> None:
        self.rolled_back += 1

    def close(self) -> None:
        self.closed += 1


@pytest.fixture
def fake_session(monkeypatch: pytest.MonkeyPatch) -> _FakeSession:
    """Патчит SessionLocal в audit-модуле, возвращает единственную сессию."""
    sess = _FakeSession()
    monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)
    # get_role детерминированно → не зависим от roles.yaml lru_cache.
    monkeypatch.setattr(audit_mod, "get_role", lambda u: "analyst")
    return sess


# ---------------------------------------------------------------------------
# classify_path
# ---------------------------------------------------------------------------


def test_classify_path_analyze() -> None:
    assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/analyze") == (
        "analyze",
        "66:41:0204016:10",
    )


def test_classify_path_forecast() -> None:
    assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/forecast") == (
        "forecast",
        "66:41:0204016:10",
    )


def test_classify_path_forecast_export_not_confused_with_forecast() -> None:
    """forecast/export должен дать action='export', НЕ 'forecast'."""
    assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/forecast/export") == (
        "export",
        "66:41:0204016:10",
    )


def test_classify_path_unmatched_returns_none() -> None:
    assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10") is None
    assert audit_mod.classify_path("/api/v1/parcels/dummy/runs") is None
    assert audit_mod.classify_path("/api/v1/analytics/dashboard") is None
    assert audit_mod.classify_path("/health") is None
    assert audit_mod.classify_path("/api/v1/admin/jobs/settings") is None
    # Похожие, но не точные пути (якорь $) не матчатся.
    assert audit_mod.classify_path("/api/v1/parcels/X/analyze/extra") is None
    assert audit_mod.classify_path("/api/v1/parcels/X/forecast/export/extra") is None


# ---------------------------------------------------------------------------
# classify_path — insight writes (#948)
# ---------------------------------------------------------------------------


def test_classify_path_insight_write_methods() -> None:
    """POST/PUT/DELETE на /insights[/{id}] → ('insight_write', None)."""
    assert audit_mod.classify_path("/api/v1/insights", "POST") == ("insight_write", None)
    assert audit_mod.classify_path("/api/v1/insights/42", "PUT") == ("insight_write", None)
    assert audit_mod.classify_path("/api/v1/insights/42", "DELETE") == ("insight_write", None)


def test_classify_path_insight_get_not_audited() -> None:
    """GET insight'ов (list/one) НЕ аудируется — это read-only."""
    assert audit_mod.classify_path("/api/v1/insights", "GET") is None
    assert audit_mod.classify_path("/api/v1/insights/42", "GET") is None


def test_classify_path_insight_without_method_skipped() -> None:
    """Без method insight-ветка не срабатывает (обратная совместимость)."""
    assert audit_mod.classify_path("/api/v1/insights") is None
    assert audit_mod.classify_path("/api/v1/insights/42") is None


def test_classify_path_insight_nested_path_not_matched() -> None:
    """Вложенные пути под /insights/{id}/... не матчатся (якорь $)."""
    assert audit_mod.classify_path("/api/v1/insights/42/comments", "POST") is None


def test_classify_path_parcels_method_ignored() -> None:
    """Parcels-паттерны не зависят от method — analyze матчится при любом методе."""
    assert audit_mod.classify_path("/api/v1/parcels/66:41:0204016:10/analyze", "GET") == (
        "analyze",
        "66:41:0204016:10",
    )


# ---------------------------------------------------------------------------
# write_audit_row
# ---------------------------------------------------------------------------


def test_write_audit_row_inserts_one_row(fake_session: _FakeSession) -> None:
    audit_mod.write_audit_row(
        username="analysttest",
        action="analyze",
        method="POST",
        path="/api/v1/parcels/66:41:0204016:10/analyze",
        status_code=200,
        cad_num="66:41:0204016:10",
    )
    assert len(fake_session.executed) == 1
    sql, params = fake_session.executed[0]
    assert "INSERT INTO audit_log" in sql
    # psycopg v3 CAST-конвенция: никаких ':param::type' в SQL.
    assert "::" not in sql
    assert params == {
        "username": "analysttest",
        "role": "analyst",
        "action": "analyze",
        "method": "POST",
        "path": "/api/v1/parcels/66:41:0204016:10/analyze",
        "cad_num": "66:41:0204016:10",
        "status_code": 200,
    }
    assert fake_session.committed == 1
    assert fake_session.closed == 1
    assert fake_session.rolled_back == 0


def test_write_audit_row_failure_is_swallowed(monkeypatch: pytest.MonkeyPatch) -> None:
    """Сбой execute НЕ должен пробрасываться — best-effort. rollback + close."""
    sess = _FakeSession(fail_on_execute=True)
    monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)
    monkeypatch.setattr(audit_mod, "get_role", lambda u: "analyst")

    # Не должно бросить.
    audit_mod.write_audit_row(
        username="analysttest",
        action="forecast",
        method="GET",
        path="/api/v1/parcels/66:41:0204016:10/forecast",
        status_code=500,
        cad_num="66:41:0204016:10",
    )
    assert sess.committed == 0
    assert sess.rolled_back == 1
    assert sess.closed == 1


def test_write_audit_row_unknown_user_role_none(monkeypatch: pytest.MonkeyPatch) -> None:
    """get_role KeyError → role=None, запись всё равно делается."""
    sess = _FakeSession()
    monkeypatch.setattr(audit_mod, "SessionLocal", lambda: sess)

    def _raise(_: str) -> str:
        raise KeyError("unknown")

    monkeypatch.setattr(audit_mod, "get_role", _raise)

    audit_mod.write_audit_row(
        username="ghost",
        action="export",
        method="GET",
        path="/api/v1/parcels/66:41:0204016:10/forecast/export",
        status_code=403,
        cad_num="66:41:0204016:10",
    )
    assert len(sess.executed) == 1
    _, params = sess.executed[0]
    assert params["role"] is None
    assert params["username"] == "ghost"


# ---------------------------------------------------------------------------
# Middleware — копия audit_log_middleware БЕЗ settings.testing-bypass (как
# rbac_guard в test_rbac.py: прод-middleware на settings.testing смотрит,
# поэтому здесь гоняем «развязанную» копию, идентичную по логике).
# ---------------------------------------------------------------------------


def _build_audited_app(captured: list[dict[str, Any]]) -> FastAPI:
    app = FastAPI()

    @app.middleware("http")
    async def audit_mw(
        request: Request,
        call_next: Callable[[Request], Awaitable[Response]],
    ) -> Response:
        matched = audit_mod.classify_path(request.url.path)
        response = await call_next(request)
        if matched is None:
            return response
        action, cad_num = matched
        username = request.headers.get("X-Authenticated-User")
        if not username:
            return response
        captured.append(
            {
                "username": username,
                "action": action,
                "method": request.method,
                "path": request.url.path,
                "status_code": response.status_code,
                "cad_num": cad_num,
            }
        )
        return response

    @app.post("/api/v1/parcels/{cad}/analyze")
    async def analyze(cad: str) -> dict:
        return {"ok": True, "cad": cad}

    @app.get("/api/v1/parcels/{cad}/forecast")
    async def forecast(cad: str) -> dict:
        return {"ok": True, "cad": cad}

    @app.get("/api/v1/parcels/{cad}/forecast/export")
    async def forecast_export(cad: str) -> dict:
        return {"ok": True, "cad": cad}

    @app.get("/api/v1/analytics/dashboard")
    async def analytics() -> dict:
        return {"ok": True}

    @app.get("/health")
    async def health() -> dict:
        return {"status": "ok"}

    return app


def test_middleware_audits_matched_request() -> None:
    captured: list[dict[str, Any]] = []
    client = TestClient(_build_audited_app(captured))

    resp = client.post(
        "/api/v1/parcels/66:41:0204016:10/analyze",
        headers={"X-Authenticated-User": "analysttest"},
    )
    assert resp.status_code == 200
    assert len(captured) == 1
    row = captured[0]
    assert row["username"] == "analysttest"
    assert row["action"] == "analyze"
    assert row["method"] == "POST"
    assert row["cad_num"] == "66:41:0204016:10"
    assert row["status_code"] == 200


def test_middleware_does_not_audit_unmatched_paths() -> None:
    captured: list[dict[str, Any]] = []
    client = TestClient(_build_audited_app(captured))

    r1 = client.get(
        "/api/v1/analytics/dashboard",
        headers={"X-Authenticated-User": "analysttest"},
    )
    r2 = client.get("/health")
    assert r1.status_code == 200
    assert r2.status_code == 200
    assert captured == []


def test_middleware_export_action_distinct_from_forecast() -> None:
    captured: list[dict[str, Any]] = []
    client = TestClient(_build_audited_app(captured))

    client.get(
        "/api/v1/parcels/66:41:0204016:10/forecast",
        headers={"X-Authenticated-User": "analysttest"},
    )
    client.get(
        "/api/v1/parcels/66:41:0204016:10/forecast/export",
        headers={"X-Authenticated-User": "analysttest"},
    )
    assert [r["action"] for r in captured] == ["forecast", "export"]


def test_middleware_audit_failure_does_not_break_request(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    """Сбой записи аудита НЕ ломает основной запрос — real response отдан.

    Здесь гоняем реальный audit_log_middleware (settings.testing выключаем
    для этого app), но write_audit_row патчим на бросающий — проверяем что
    клиент всё равно получает 200 от handler'а."""
    from app.core.config import settings

    monkeypatch.setattr(settings, "testing", False)

    def _boom(**_: Any) -> None:
        raise RuntimeError("audit exploded")

    monkeypatch.setattr(audit_mod, "write_audit_row", _boom)

    app = FastAPI()
    app.middleware("http")(audit_mod.audit_log_middleware)

    @app.post("/api/v1/parcels/{cad}/analyze")
    async def analyze(cad: str) -> dict:
        return {"ok": True, "cad": cad}

    client = TestClient(app)
    resp = client.post(
        "/api/v1/parcels/66:41:0204016:10/analyze",
        headers={"X-Authenticated-User": "analysttest"},
    )
    # write_audit_row взорвался, но middleware проглотил → handler-ответ дошёл.
    assert resp.status_code == 200
    assert resp.json()["ok"] is True