82928be326
All checks were successful
CI Trade-In / backend-tests (pull_request) Successful in 59s
CI / changes (pull_request) Successful in 9s
CI Trade-In / changes (pull_request) Successful in 10s
CI Trade-In / frontend-checks (pull_request) Has been skipped
CI / backend-tests (pull_request) Has been skipped
CI / frontend-tests (pull_request) Has been skipped
CI / openapi-codegen-check (pull_request) Has been skipped
Server-side capture layer for Feature 2 (login/IP audit) + base for Feature 3 (behavior analytics), on top of the already-deployed user_events table (migration 184). record_event/schedule_event never raise into the caller — audit logging must never break a real request.
116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
"""Tests for RequestAuditMiddleware (app/core/request_audit.py).
|
|
|
|
Тестируем middleware в изоляции на минимальном FastAPI-приложении (как
|
|
tests/test_ratelimit.py) — не тянем тяжёлый app.main. schedule_event
|
|
и should_log_login мокируются, реальная запись в user_events не требуется
|
|
(она уже покрыта tests/test_user_events.py).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
|
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.core.request_audit import RequestAuditMiddleware
|
|
|
|
|
|
@pytest.fixture
|
|
def client() -> TestClient:
|
|
app = FastAPI()
|
|
app.add_middleware(RequestAuditMiddleware)
|
|
|
|
@app.get("/api/v1/ping")
|
|
def ping() -> dict[str, bool]:
|
|
return {"ok": True}
|
|
|
|
@app.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
return TestClient(app)
|
|
|
|
|
|
def test_authenticated_api_request_schedules_api_request_event(client: TestClient) -> None:
|
|
with (
|
|
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
|
patch("app.core.request_audit.should_log_login", return_value=False),
|
|
):
|
|
resp = client.get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
|
|
|
assert resp.status_code == 200
|
|
mock_schedule.assert_called_once()
|
|
kwargs = mock_schedule.call_args.kwargs
|
|
assert kwargs["event_type"] == "api_request"
|
|
assert kwargs["username"] == "alice"
|
|
assert kwargs["path"] == "/api/v1/ping"
|
|
assert kwargs["method"] == "GET"
|
|
|
|
|
|
def test_login_event_scheduled_when_should_log_login_true(client: TestClient) -> None:
|
|
with (
|
|
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
|
patch("app.core.request_audit.should_log_login", return_value=True),
|
|
):
|
|
client.get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
|
|
|
event_types = [c.kwargs["event_type"] for c in mock_schedule.call_args_list]
|
|
assert event_types == ["api_request", "login"]
|
|
|
|
|
|
def test_login_event_not_scheduled_when_should_log_login_false(client: TestClient) -> None:
|
|
with (
|
|
patch("app.core.request_audit.schedule_event") as mock_schedule,
|
|
patch("app.core.request_audit.should_log_login", return_value=False),
|
|
):
|
|
client.get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
|
|
|
event_types = [c.kwargs["event_type"] for c in mock_schedule.call_args_list]
|
|
assert event_types == ["api_request"]
|
|
|
|
|
|
def test_no_username_header_skips_audit(client: TestClient) -> None:
|
|
with patch("app.core.request_audit.schedule_event") as mock_schedule:
|
|
resp = client.get("/api/v1/ping")
|
|
|
|
assert resp.status_code == 200
|
|
mock_schedule.assert_not_called()
|
|
|
|
|
|
def test_public_path_skips_audit_even_with_username(client: TestClient) -> None:
|
|
with patch("app.core.request_audit.schedule_event") as mock_schedule:
|
|
resp = client.get("/health", headers={"X-Authenticated-User": "alice"})
|
|
|
|
assert resp.status_code == 200
|
|
mock_schedule.assert_not_called()
|
|
|
|
|
|
def test_non_api_path_skips_audit(client: TestClient) -> None:
|
|
"""Path not under /api/ (and not one of the FastAPI-generated docs paths) is skipped."""
|
|
app = FastAPI()
|
|
app.add_middleware(RequestAuditMiddleware)
|
|
|
|
@app.get("/other")
|
|
def other() -> dict[str, bool]:
|
|
return {"ok": True}
|
|
|
|
with patch("app.core.request_audit.schedule_event") as mock_schedule:
|
|
resp = TestClient(app).get("/other", headers={"X-Authenticated-User": "alice"})
|
|
|
|
assert resp.status_code == 200
|
|
mock_schedule.assert_not_called()
|
|
|
|
|
|
def test_audit_failure_does_not_break_response(client: TestClient) -> None:
|
|
"""A raising schedule_event must never turn a 200 into a 500 (dispatch swallows it)."""
|
|
with patch("app.core.request_audit.schedule_event", side_effect=RuntimeError("boom")):
|
|
resp = client.get("/api/v1/ping", headers={"X-Authenticated-User": "alice"})
|
|
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"ok": True}
|