gendesign/tradein-mvp/backend/tests/test_audit_api.py

"""Tests for app.api.v1.audit — admin read API over `user_events` (Feature 2/3).

Coverage:
  (a) static SQL guard — CAST(:x AS type), never `:x::type` (psycopg v3 trap).
  (b) GET /audit/accounts — empty table → [] (never errors); populated → shape-valid
      AccountSummary rows pass through response_model unchanged.
  (c) GET /audit/accounts/{username} — empty table (incl. unknown username, NOT 404) →
      all 4 lists == []; populated → shape-valid drilldown rows.
  (d) GET /analytics — empty table → zeroed summary + empty lists (never errors);
      populated → shape-valid dashboard bundle, `days` query param clamped [1, 365].
  (e) optional real-Postgres round trip — inserts a couple of user_events rows and
      asserts aggregation through the real DB; self-SKIPS without a reachable,
      non-placeholder Postgres (mirrors tests/test_house_dedup_merge.py's live-DB
      pattern).

Router is tested in isolation on a minimal FastAPI app (mirrors tests/test_ratelimit.py
and the quota_app fixture in tests/test_user_events.py) — no need to pull in the full
app.main (sentry/scheduler/CORS/rate-limit wiring).
"""

from __future__ import annotations

import inspect
import os
import re
import uuid
from datetime import UTC, date, datetime
from typing import Any

# psycopg v3 driver required; stub DATABASE_URL before any app import (mirrors other tests).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api.v1 import audit as audit_module
from app.core.db import get_db

# ---------------------------------------------------------------------------
# (a) Static SQL guard
# ---------------------------------------------------------------------------

_AUDIT_SRC = inspect.getsource(audit_module)


def test_no_psycopg_v3_doublecolon_cast() -> None:
    """CAST(:x AS type) — НИКОГДА `:x::type` (psycopg v3 trap, .claude/rules/backend.md)."""
    assert not re.search(r":\w+::\w", _AUDIT_SRC)


def test_days_param_uses_cast_as_int() -> None:
    assert "CAST(:days AS int)" in _AUDIT_SRC


# ---------------------------------------------------------------------------
# Fakes — mirror the mocked-DB convention used across tests/test_user_events.py etc.
# ---------------------------------------------------------------------------


class _FakeMappingResult:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self._rows = rows

    def all(self) -> list[dict[str, Any]]:
        return list(self._rows)

    def one(self) -> dict[str, Any]:
        return self._rows[0]


class _FakeExecResult:
    def __init__(self, rows: list[dict[str, Any]]) -> None:
        self._rows = rows

    def mappings(self) -> _FakeMappingResult:
        return _FakeMappingResult(self._rows)


class _FakeSession:
    """Returns queued row-lists in the exact order audit.py issues db.execute() calls."""

    def __init__(self, responses: list[list[dict[str, Any]]]) -> None:
        self._responses = list(responses)
        self._i = 0

    def execute(self, _stmt: object, _params: dict[str, Any] | None = None) -> _FakeExecResult:
        rows = self._responses[self._i]
        self._i += 1
        return _FakeExecResult(rows)


def _make_app(responses: list[list[dict[str, Any]]]) -> FastAPI:
    application = FastAPI()
    application.include_router(audit_module.router, prefix="/api/v1/admin")

    def _override_db() -> Any:
        yield _FakeSession(responses)

    application.dependency_overrides[get_db] = _override_db
    return application


_NOW = datetime(2026, 7, 13, 12, 0, 0, tzinfo=UTC)

# ---------------------------------------------------------------------------
# (b) GET /audit/accounts
# ---------------------------------------------------------------------------


def test_list_accounts_empty_table() -> None:
    app = _make_app(responses=[[]])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/audit/accounts")
    assert resp.status_code == 200
    assert resp.json() == []


def test_list_accounts_populated_shape() -> None:
    row = {
        "username": "user1",
        "first_seen_at": _NOW,
        "last_seen_at": _NOW,
        "distinct_ips": 3,
        "distinct_devices": 2,
        "login_count": 5,
        "request_count": 40,
        "search_count": 7,
    }
    app = _make_app(responses=[[row]])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/audit/accounts")
    assert resp.status_code == 200
    data = resp.json()
    assert len(data) == 1
    assert data[0]["username"] == "user1"
    assert data[0]["distinct_ips"] == 3
    assert data[0]["login_count"] == 5
    assert data[0]["request_count"] == 40
    assert data[0]["search_count"] == 7


# ---------------------------------------------------------------------------
# (c) GET /audit/accounts/{username}
# ---------------------------------------------------------------------------


def test_account_drilldown_empty_table_unknown_user_not_404() -> None:
    """Unknown username → 200 with all 4 lists empty, NOT a 404."""
    app = _make_app(responses=[[], [], [], []])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/audit/accounts/ghost_user_xyz")
    assert resp.status_code == 200
    data = resp.json()
    assert data == {"ips": [], "devices": [], "searches": [], "recent_activity": []}


def test_account_drilldown_populated_shape() -> None:
    ips = [{"ip_address": "1.2.3.4", "event_count": 10, "first_seen": _NOW, "last_seen": _NOW}]
    devices = [
        {"user_agent": "pytest-agent", "event_count": 10, "first_seen": _NOW, "last_seen": _NOW}
    ]
    searches = [
        {
            "address": "ул. Малышева, 1",
            "area_m2": "50.0",
            "rooms": "2",
            "estimate_id": str(uuid.uuid4()),
            "ip_address": "1.2.3.4",
            "created_at": _NOW,
        }
    ]
    recent_activity = [
        {
            "event_type": "api_request",
            "path": "/api/v1/trade-in/estimate",
            "method": "POST",
            "ip_address": "1.2.3.4",
            "created_at": _NOW,
        }
    ]
    app = _make_app(responses=[ips, devices, searches, recent_activity])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/audit/accounts/user1")
    assert resp.status_code == 200
    data = resp.json()
    assert data["ips"][0]["ip_address"] == "1.2.3.4"
    assert data["ips"][0]["event_count"] == 10
    assert data["devices"][0]["user_agent"] == "pytest-agent"
    assert data["searches"][0]["address"] == "ул. Малышева, 1"
    assert data["searches"][0]["area_m2"] == "50.0"
    assert data["recent_activity"][0]["event_type"] == "api_request"
    assert data["recent_activity"][0]["path"] == "/api/v1/trade-in/estimate"


# ---------------------------------------------------------------------------
# (d) GET /analytics
# ---------------------------------------------------------------------------

_ZERO_SUMMARY = {
    "total_events": 0,
    "distinct_users": 0,
    "events_last_24h": 0,
    "active_users_last_24h": 0,
}


def test_analytics_empty_table_never_errors() -> None:
    app = _make_app(responses=[[_ZERO_SUMMARY], [], [], [], []])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/analytics")
    assert resp.status_code == 200
    data = resp.json()
    assert data["summary"] == _ZERO_SUMMARY
    assert data["daily"] == []
    assert data["top_searches"] == []
    assert data["top_paths"] == []
    assert data["by_account"] == []


def test_analytics_populated_shape() -> None:
    summary = {
        "total_events": 120,
        "distinct_users": 4,
        "events_last_24h": 15,
        "active_users_last_24h": 2,
    }
    daily = [{"day": date(2026, 7, 13), "events": 15, "users": 2}]
    top_searches = [{"address": "ул. Малышева, 1", "n": 3}]
    top_paths = [{"path": "/api/v1/trade-in/estimate", "n": 40}]
    by_account = [{"username": "user1", "events": 60, "searches": 7, "last_seen": _NOW}]
    app = _make_app(responses=[[summary], daily, top_searches, top_paths, by_account])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/analytics")
    assert resp.status_code == 200
    data = resp.json()
    assert data["summary"]["total_events"] == 120
    assert data["daily"][0]["events"] == 15
    assert data["daily"][0]["day"] == "2026-07-13"
    assert data["top_searches"][0]["address"] == "ул. Малышева, 1"
    assert data["top_paths"][0]["n"] == 40
    assert data["by_account"][0]["username"] == "user1"


def test_analytics_days_query_param_clamped() -> None:
    """days=0 (below ge=1) → 422; days=9999 (above le=365) → 422."""
    app = _make_app(responses=[[_ZERO_SUMMARY], [], [], [], []])
    client = TestClient(app)
    assert client.get("/api/v1/admin/analytics?days=0").status_code == 422
    assert client.get("/api/v1/admin/analytics?days=9999").status_code == 422


def test_analytics_days_query_param_default_30() -> None:
    app = _make_app(responses=[[_ZERO_SUMMARY], [], [], [], []])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/analytics")
    assert resp.status_code == 200


def test_analytics_days_query_param_accepted_in_range() -> None:
    app = _make_app(responses=[[_ZERO_SUMMARY], [], [], [], []])
    client = TestClient(app)
    resp = client.get("/api/v1/admin/analytics?days=7")
    assert resp.status_code == 200


# ---------------------------------------------------------------------------
# (e) Optional real-Postgres round trip (self-skips without a reachable DB)
# ---------------------------------------------------------------------------


def _live_session() -> Any | None:
    """Return a SQLAlchemy Session if a non-placeholder Postgres is reachable, else None."""
    try:
        from sqlalchemy import create_engine
        from sqlalchemy import text as _t
        from sqlalchemy.orm import sessionmaker

        dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "")
        if not dsn or "localhost:5432/test" in dsn:
            return None
        engine = create_engine(dsn, future=True)
        conn = engine.connect()
        conn.execute(_t("SELECT 1"))
        conn.close()
        return sessionmaker(bind=engine, future=True)()
    except Exception:
        return None


@pytest.mark.skipif(_live_session() is None, reason="no reachable Postgres test DB")
def test_real_accounts_and_analytics_aggregate_inserted_rows() -> None:
    """End-to-end on a real DB: insert 2 user_events rows for a throwaway test username,
    assert /audit/accounts, /audit/accounts/{username} and /analytics reflect them."""
    from sqlalchemy import text as _t

    from app.core.db import SessionLocal

    db = _live_session()
    assert db is not None
    username = f"audit_test_{uuid.uuid4().hex[:8]}"
    try:
        db.execute(
            _t(
                """
                INSERT INTO user_events
                    (event_type, username, ip_address, user_agent, path, method, payload)
                VALUES
                    ('login', :u, CAST('9.9.9.9' AS inet), 'pytest-ua', NULL, NULL,
                     CAST('{}' AS jsonb)),
                    ('estimate_request', :u, CAST('9.9.9.9' AS inet), 'pytest-ua',
                     '/api/v1/trade-in/estimate', 'POST',
                     CAST(:payload AS jsonb))
                """
            ),
            {"u": username, "payload": '{"address": "ул. Тестовая, 1", "area_m2": "42.0"}'},
        )
        db.commit()

        application = FastAPI()
        application.include_router(audit_module.router, prefix="/api/v1/admin")

        def _override_db() -> Any:
            yield SessionLocal()

        application.dependency_overrides[get_db] = _override_db
        client = TestClient(application)

        accounts = client.get("/api/v1/admin/audit/accounts").json()
        row = next(r for r in accounts if r["username"] == username)
        assert row["login_count"] == 1
        assert row["search_count"] == 1

        drilldown = client.get(f"/api/v1/admin/audit/accounts/{username}").json()
        assert len(drilldown["searches"]) == 1
        assert drilldown["searches"][0]["address"] == "ул. Тестовая, 1"
        assert len(drilldown["recent_activity"]) == 2

        dashboard = client.get("/api/v1/admin/analytics?days=1").json()
        assert dashboard["summary"]["total_events"] >= 2
    finally:
        db.rollback()
        db.execute(_t("DELETE FROM user_events WHERE username = :u"), {"u": username})
        db.commit()