feat(tradein/audit): request-audit middleware + estimate_request logging → user_events #2519
6 changed files with 645 additions and 1 deletions
|
|
@ -11,12 +11,13 @@ from datetime import UTC, date, datetime, timedelta
|
|||
from typing import Annotated, Any
|
||||
from uuid import UUID
|
||||
|
||||
from fastapi import APIRouter, Depends, File, Header, HTTPException, Response, UploadFile
|
||||
from fastapi import APIRouter, Depends, File, Header, HTTPException, Request, Response, UploadFile
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.core.config import settings
|
||||
from app.core.db import get_db
|
||||
from app.core.ratelimit import _client_ip
|
||||
from app.schemas.trade_in import (
|
||||
AggregatedEstimate,
|
||||
AnalogLot,
|
||||
|
|
@ -45,6 +46,7 @@ from app.schemas.trade_in import (
|
|||
from app.services import account_quota
|
||||
from app.services.exporters.trade_in_pdf import generate_trade_in_pdf
|
||||
from app.services.image_sanitizer import ImageSanitizationError, sanitize_image
|
||||
from app.services.user_events import schedule_event
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
|
@ -147,6 +149,7 @@ def _resolve_target_house_id(
|
|||
@router.post("/estimate", response_model=AggregatedEstimate)
|
||||
async def estimate(
|
||||
payload: TradeInEstimateInput,
|
||||
request: Request,
|
||||
db: Annotated[Session, Depends(get_db)],
|
||||
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
|
||||
) -> AggregatedEstimate:
|
||||
|
|
@ -183,6 +186,24 @@ async def estimate(
|
|||
# тут: при гонке двух /estimate на used=MONTHLY_LIMIT-1 второй получит False.
|
||||
if not account_quota.increment(db, x_authenticated_user):
|
||||
raise HTTPException(status_code=429, detail=account_quota.LIMIT_EXHAUSTED_MESSAGE)
|
||||
|
||||
# Feature 2/3 foundation: "что искали" — обогащённая estimate_request-запись
|
||||
# в user_events (адрес/площадь/комнаты + estimate_id для join с trade_in_estimates).
|
||||
# schedule_event сам никогда не raises — сбой аудита не должен ронять ответ.
|
||||
schedule_event(
|
||||
event_type="estimate_request",
|
||||
username=x_authenticated_user or "",
|
||||
ip=_client_ip(request),
|
||||
user_agent=request.headers.get("user-agent"),
|
||||
path=str(request.url.path),
|
||||
method="POST",
|
||||
estimate_id=str(result.estimate_id),
|
||||
payload={
|
||||
"address": payload.address,
|
||||
"area_m2": payload.area_m2,
|
||||
"rooms": payload.rooms,
|
||||
},
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
|
|
|
|||
66
tradein-mvp/backend/app/core/request_audit.py
Normal file
66
tradein-mvp/backend/app/core/request_audit.py
Normal file
|
|
@ -0,0 +1,66 @@
|
|||
"""RequestAuditMiddleware — пишет `api_request` (+ дедуплицированный `login`)
|
||||
события в `user_events` для каждого аутентифицированного `/api/*` запроса.
|
||||
|
||||
Foundation для Feature 2 (login/IP audit) и базы Feature 3 (behavior analytics).
|
||||
Логирование выполняется ПОСЛЕ `call_next` (не задерживает и не ветвит реальный
|
||||
ответ клиенту) и через fire-and-forget `schedule_event` — сбой аудита никогда
|
||||
не влияет на HTTP-ответ.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
|
||||
from fastapi import Request
|
||||
from starlette.middleware.base import BaseHTTPMiddleware
|
||||
from starlette.responses import Response
|
||||
|
||||
from app.core.ratelimit import _client_ip
|
||||
from app.services.user_events import schedule_event, should_log_login
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Зеркалит app.main._PUBLIC_PATHS. Не импортируем напрямую из app.main — оно
|
||||
# импортирует этот модуль (регистрирует middleware), обратный импорт дал бы
|
||||
# циклическую зависимость.
|
||||
_PUBLIC_PATHS = frozenset({"/health", "/docs", "/redoc", "/openapi.json"})
|
||||
|
||||
|
||||
class RequestAuditMiddleware(BaseHTTPMiddleware):
|
||||
"""Логирует активность аутентифицированных пользователей в `user_events`."""
|
||||
|
||||
async def dispatch(self, request: Request, call_next): # type: ignore[no-untyped-def]
|
||||
response: Response = await call_next(request)
|
||||
|
||||
try:
|
||||
username = request.headers.get("x-authenticated-user")
|
||||
path = request.url.path
|
||||
if username and path.startswith("/api/") and path not in _PUBLIC_PATHS:
|
||||
ip = _client_ip(request)
|
||||
ua = request.headers.get("user-agent")
|
||||
|
||||
# Общий behavior/activity-поток — каждый authenticated API-запрос.
|
||||
schedule_event(
|
||||
event_type="api_request",
|
||||
username=username,
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
path=path,
|
||||
method=request.method,
|
||||
)
|
||||
|
||||
# Дедуплицированный login/IP-audit сигнал — максимум раз в день
|
||||
# на (юзер, IP, устройство).
|
||||
if should_log_login(username, ip, ua):
|
||||
schedule_event(
|
||||
event_type="login",
|
||||
username=username,
|
||||
ip=ip,
|
||||
user_agent=ua,
|
||||
path=path,
|
||||
method=request.method,
|
||||
)
|
||||
except Exception:
|
||||
logger.warning("RequestAuditMiddleware: failed to record event", exc_info=True)
|
||||
|
||||
return response
|
||||
|
|
@ -29,6 +29,7 @@ from app.core.config import settings
|
|||
from app.core.db import SessionLocal
|
||||
from app.core.fdw import ensure_fdw_user_mapping
|
||||
from app.core.ratelimit import RateLimitMiddleware
|
||||
from app.core.request_audit import RequestAuditMiddleware
|
||||
from app.observability.sentry_scrub import scrub_pii_event
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
|
@ -213,6 +214,8 @@ app.add_middleware(
|
|||
)
|
||||
# Rate-limit публичного API (per-user / per-IP sliding window) — защита от абуза.
|
||||
app.add_middleware(RateLimitMiddleware)
|
||||
# Request-audit: пишет api_request/login события в user_events (Feature 2/3 foundation).
|
||||
app.add_middleware(RequestAuditMiddleware)
|
||||
|
||||
|
||||
@app.get("/health")
|
||||
|
|
|
|||
146
tradein-mvp/backend/app/services/user_events.py
Normal file
146
tradein-mvp/backend/app/services/user_events.py
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
"""Сервис записи в `user_events` — Feature 2 (login/IP audit) + база для Feature 3
|
||||
(behavior analytics).
|
||||
|
||||
`user_events` (migration `184_user_events.sql`) — unified append-only event log,
|
||||
admin-read-only. Запись события НИКОГДА не должна ронять реальный HTTP-запрос:
|
||||
любая ошибка (БД недоступна, сетевой сбой, схема разъехалась) ловится и логируется
|
||||
через `logger.warning`, без re-raise — вызывающий код (middleware / handler)
|
||||
продолжает работать так, как если бы аудит был выключен.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from datetime import UTC, datetime
|
||||
from typing import Any
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def record_event(
|
||||
*,
|
||||
event_type: str,
|
||||
username: str,
|
||||
ip: str | None = None,
|
||||
user_agent: str | None = None,
|
||||
path: str | None = None,
|
||||
method: str | None = None,
|
||||
estimate_id: str | None = None,
|
||||
payload: dict[str, Any] | None = None,
|
||||
) -> None:
|
||||
"""Синхронно вставляет одну строку в `user_events`.
|
||||
|
||||
Открывает СОБСТВЕННУЮ сессию (`SessionLocal()`), декаплённую от транзакции
|
||||
вызывающего запроса — событие коммитится независимо и переживает rollback
|
||||
основного хендлера (и наоборот: сбой записи события не трогает основную tx,
|
||||
т.к. она уже закоммичена/не связана с этой сессией).
|
||||
|
||||
Никогда не поднимает исключение — вызывающий код (middleware / endpoint)
|
||||
не должен падать из-за проблем с аудит-логом.
|
||||
"""
|
||||
try:
|
||||
with SessionLocal() as db:
|
||||
db.execute(
|
||||
text(
|
||||
"""
|
||||
INSERT INTO user_events
|
||||
(event_type, username, ip_address, user_agent, path, method,
|
||||
estimate_id, payload)
|
||||
VALUES
|
||||
(:event_type, :username, CAST(:ip AS inet), :user_agent, :path,
|
||||
:method, CAST(:estimate_id AS uuid), CAST(:payload AS jsonb))
|
||||
"""
|
||||
),
|
||||
{
|
||||
"event_type": event_type,
|
||||
"username": username,
|
||||
"ip": ip,
|
||||
"user_agent": user_agent,
|
||||
"path": path,
|
||||
"method": method,
|
||||
"estimate_id": estimate_id,
|
||||
"payload": json.dumps(payload or {}, ensure_ascii=False),
|
||||
},
|
||||
)
|
||||
db.commit()
|
||||
except Exception:
|
||||
logger.warning(
|
||||
"user_events: failed to record event_type=%r username=%r path=%r",
|
||||
event_type,
|
||||
username,
|
||||
path,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
|
||||
def schedule_event(**kwargs: Any) -> None:
|
||||
"""Fire-and-forget обёртка над `record_event` — никогда не блокирует запрос и
|
||||
никогда не поднимает исключение наружу.
|
||||
|
||||
- Есть running event loop (обычный async FastAPI handler / middleware) →
|
||||
INSERT уводится в отдельный поток через `asyncio.to_thread`, вызывающая
|
||||
корутина не ждёт результата. Ошибки ловятся done-callback'ом и просто
|
||||
логируются (иначе asyncio ругался бы "Task exception was never retrieved").
|
||||
- Нет running loop (sync-контекст: тесты, Celery task, скрипт) → вызывает
|
||||
`record_event(**kwargs)` синхронно inline (она сама никогда не raises).
|
||||
"""
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
record_event(**kwargs)
|
||||
return
|
||||
|
||||
try:
|
||||
task = loop.create_task(asyncio.to_thread(record_event, **kwargs))
|
||||
except Exception:
|
||||
logger.warning("user_events: failed to schedule background event", exc_info=True)
|
||||
return
|
||||
|
||||
def _log_task_exception(t: asyncio.Task[None]) -> None:
|
||||
if t.cancelled():
|
||||
return
|
||||
exc = t.exception()
|
||||
if exc is not None:
|
||||
logger.warning("user_events: background record_event failed: %s", exc)
|
||||
|
||||
task.add_done_callback(_log_task_exception)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# In-process best-effort dedup для login-события.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_LOGIN_DEDUP_LOCK = threading.Lock()
|
||||
_LOGIN_DEDUP_SEEN: set[str] = set()
|
||||
_LOGIN_DEDUP_MAX = 50_000
|
||||
|
||||
|
||||
def should_log_login(username: str, ip: str | None, user_agent: str | None) -> bool:
|
||||
"""True максимум один раз в сутки на комбинацию (username, ip, user_agent) в
|
||||
рамках жизни процесса — держит объём `event_type='login'` строк на уровне
|
||||
~одной записи на аккаунт+устройство в день, а не одной на каждый API-запрос.
|
||||
|
||||
Best-effort/не персистентно: рестарт процесса (деплой, worker respawn)
|
||||
сбрасывает in-memory set, так что после рестарта первая комбинация дня
|
||||
залогируется заново — недо-дедуп, не потеря данных (для отчётов по
|
||||
user_events всё равно используется `DISTINCT (username, ip_address,
|
||||
user_agent, date_trunc('day', created_at))`, так что дубликаты не искажают
|
||||
метрики, лишь чуть увеличивают объём записи).
|
||||
"""
|
||||
day = datetime.now(UTC).strftime("%Y-%m-%d")
|
||||
key = f"{username}|{ip or ''}|{user_agent or ''}|{day}"
|
||||
|
||||
with _LOGIN_DEDUP_LOCK:
|
||||
if key in _LOGIN_DEDUP_SEEN:
|
||||
return False
|
||||
if len(_LOGIN_DEDUP_SEEN) >= _LOGIN_DEDUP_MAX:
|
||||
_LOGIN_DEDUP_SEEN.clear()
|
||||
_LOGIN_DEDUP_SEEN.add(key)
|
||||
return True
|
||||
116
tradein-mvp/backend/tests/test_request_audit.py
Normal file
116
tradein-mvp/backend/tests/test_request_audit.py
Normal file
|
|
@ -0,0 +1,116 @@
|
|||
"""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}
|
||||
292
tradein-mvp/backend/tests/test_user_events.py
Normal file
292
tradein-mvp/backend/tests/test_user_events.py
Normal file
|
|
@ -0,0 +1,292 @@
|
|||
"""Tests for app.services.user_events — record_event / schedule_event / should_log_login.
|
||||
|
||||
Coverage:
|
||||
(a) record_event inserts exactly one row via a mocked SessionLocal, commits, and the
|
||||
SQL uses CAST(:x AS type) (never the psycopg-v3-broken `:x::type` form).
|
||||
(b) record_event swallows exceptions — a DB failure (connect or execute) never raises.
|
||||
(c) schedule_event in a sync context (no running event loop) calls record_event inline.
|
||||
(d) schedule_event in an async context schedules a non-blocking background task that
|
||||
still calls through, and a failure inside it is logged, not raised.
|
||||
(e) should_log_login: True once per (user, ip, ua, day), False on repeat; independent
|
||||
per user / ip / user_agent; prunes the in-process set past capacity.
|
||||
(f) optional real-DB round trip against app.core.db.SessionLocal (the exact session
|
||||
record_event uses) — self-skips without a reachable, non-placeholder Postgres,
|
||||
mirroring tests/test_house_dedup_merge.py's live-DB pattern.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import os
|
||||
import re
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
# 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 app.services import user_events as ue
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fakes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeResult:
|
||||
def fetchone(self) -> None:
|
||||
return None
|
||||
|
||||
|
||||
class _FakeSession:
|
||||
"""SessionLocal() stand-in supporting the `with SessionLocal() as db:` pattern."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.executed: list[tuple[str, dict[str, Any] | None]] = []
|
||||
self.committed = False
|
||||
|
||||
def execute(self, stmt: object, params: dict[str, Any] | None = None) -> _FakeResult:
|
||||
self.executed.append((str(stmt), params))
|
||||
return _FakeResult()
|
||||
|
||||
def commit(self) -> None:
|
||||
self.committed = True
|
||||
|
||||
def __enter__(self) -> _FakeSession:
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc: object) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# (a) record_event — successful insert
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_record_event_inserts_and_commits() -> None:
|
||||
fake = _FakeSession()
|
||||
with patch.object(ue, "SessionLocal", return_value=fake):
|
||||
ue.record_event(
|
||||
event_type="estimate_request",
|
||||
username="user1",
|
||||
ip="1.2.3.4",
|
||||
user_agent="pytest-agent",
|
||||
path="/api/v1/trade-in/estimate",
|
||||
method="POST",
|
||||
estimate_id="11111111-1111-1111-1111-111111111111",
|
||||
payload={"address": "ул. Ленина, 1", "area_m2": 50.0, "rooms": 2},
|
||||
)
|
||||
|
||||
assert fake.committed is True
|
||||
assert len(fake.executed) == 1
|
||||
sql, params = fake.executed[0]
|
||||
assert params is not None
|
||||
assert "INSERT INTO user_events" in sql
|
||||
assert params["event_type"] == "estimate_request"
|
||||
assert params["username"] == "user1"
|
||||
assert params["ip"] == "1.2.3.4"
|
||||
assert params["estimate_id"] == "11111111-1111-1111-1111-111111111111"
|
||||
# payload сериализован в JSON-строку (CAST(:payload AS jsonb) на стороне SQL).
|
||||
assert isinstance(params["payload"], str)
|
||||
assert "Ленина" in params["payload"]
|
||||
|
||||
|
||||
def test_record_event_cast_not_doublecolon() -> None:
|
||||
"""CAST(:x AS type) — НИКОГДА `:x::type` (psycopg v3 trap, .claude/rules/backend.md)."""
|
||||
fake = _FakeSession()
|
||||
with patch.object(ue, "SessionLocal", return_value=fake):
|
||||
ue.record_event(event_type="login", username="user1")
|
||||
|
||||
sql, _ = fake.executed[0]
|
||||
assert not re.search(r":\w+::\w", sql)
|
||||
assert "CAST(:ip AS inet)" in sql
|
||||
assert "CAST(:estimate_id AS uuid)" in sql
|
||||
assert "CAST(:payload AS jsonb)" in sql
|
||||
|
||||
|
||||
def test_record_event_default_payload_is_empty_dict() -> None:
|
||||
fake = _FakeSession()
|
||||
with patch.object(ue, "SessionLocal", return_value=fake):
|
||||
ue.record_event(event_type="login", username="user2")
|
||||
|
||||
_, params = fake.executed[0]
|
||||
assert params is not None
|
||||
assert params["payload"] == "{}"
|
||||
assert params["ip"] is None
|
||||
assert params["estimate_id"] is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# (b) record_event never raises
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_record_event_swallows_session_open_errors() -> None:
|
||||
class _BrokenSession:
|
||||
def __enter__(self) -> _BrokenSession:
|
||||
raise RuntimeError("db unreachable")
|
||||
|
||||
def __exit__(self, *exc: object) -> bool:
|
||||
return False
|
||||
|
||||
with patch.object(ue, "SessionLocal", return_value=_BrokenSession()):
|
||||
# Should NOT raise.
|
||||
ue.record_event(event_type="login", username="user1")
|
||||
|
||||
|
||||
def test_record_event_swallows_execute_errors() -> None:
|
||||
class _ExplodingSession(_FakeSession):
|
||||
def execute(self, stmt: object, params: dict[str, Any] | None = None) -> _FakeResult:
|
||||
raise RuntimeError("syntax error at or near ':'")
|
||||
|
||||
with patch.object(ue, "SessionLocal", return_value=_ExplodingSession()):
|
||||
ue.record_event(event_type="login", username="user1")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# (c)/(d) schedule_event
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_schedule_event_sync_context_calls_record_event_inline() -> None:
|
||||
"""No running loop (plain sync call, e.g. from a script/test) → inline call."""
|
||||
with patch.object(ue, "record_event") as mock_record:
|
||||
ue.schedule_event(event_type="login", username="user1", ip="9.9.9.9")
|
||||
|
||||
mock_record.assert_called_once_with(event_type="login", username="user1", ip="9.9.9.9")
|
||||
|
||||
|
||||
async def test_schedule_event_async_context_is_non_blocking_and_calls_through() -> None:
|
||||
"""Running loop present → backgrounded via asyncio.to_thread; does not block the
|
||||
calling coroutine and eventually calls record_event."""
|
||||
with patch.object(ue, "record_event") as mock_record:
|
||||
ue.schedule_event(event_type="api_request", username="user1", path="/api/v1/x")
|
||||
# schedule_event must return immediately without awaiting the background task.
|
||||
# Give the scheduled task a chance to run before asserting.
|
||||
await asyncio.sleep(0.05)
|
||||
|
||||
mock_record.assert_called_once_with(
|
||||
event_type="api_request", username="user1", path="/api/v1/x"
|
||||
)
|
||||
|
||||
|
||||
async def test_schedule_event_background_failure_is_logged_not_raised() -> None:
|
||||
"""A record_event failure inside the background task must not propagate/crash the
|
||||
caller — the done-callback swallows it (logs a warning)."""
|
||||
with patch.object(ue, "record_event", side_effect=RuntimeError("boom")):
|
||||
# Must not raise synchronously (the task runs in the background).
|
||||
ue.schedule_event(event_type="login", username="user1")
|
||||
await asyncio.sleep(0.05)
|
||||
# Reaching here without an unhandled-exception bubbling up confirms the
|
||||
# done-callback caught it.
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# (e) should_log_login dedup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_should_log_login_true_once_then_false_same_day() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is False
|
||||
|
||||
|
||||
def test_should_log_login_independent_per_ip() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
|
||||
assert ue.should_log_login("alice", "2.2.2.2", "chrome") is True
|
||||
|
||||
|
||||
def test_should_log_login_independent_per_user() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
|
||||
assert ue.should_log_login("bob", "1.1.1.1", "chrome") is True
|
||||
|
||||
|
||||
def test_should_log_login_independent_per_user_agent() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "firefox") is True
|
||||
|
||||
|
||||
def test_should_log_login_none_ip_and_ua_are_valid_keys() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
assert ue.should_log_login("alice", None, None) is True
|
||||
assert ue.should_log_login("alice", None, None) is False
|
||||
|
||||
|
||||
def test_should_log_login_prunes_when_over_capacity() -> None:
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
for i in range(ue._LOGIN_DEDUP_MAX):
|
||||
ue._LOGIN_DEDUP_SEEN.add(f"filler-{i}")
|
||||
assert len(ue._LOGIN_DEDUP_SEEN) == ue._LOGIN_DEDUP_MAX
|
||||
|
||||
# Next call must prune (clear the whole set) before adding, not grow unbounded.
|
||||
assert ue.should_log_login("alice", "1.1.1.1", "chrome") is True
|
||||
assert len(ue._LOGIN_DEDUP_SEEN) == 1
|
||||
ue._LOGIN_DEDUP_SEEN.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# (f) optional real-DB round trip (self-skips without a reachable test DB)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _db_reachable() -> bool:
|
||||
"""True if app.core.db's engine (the one record_event actually uses) points at a
|
||||
real, non-placeholder Postgres and is reachable — mirrors the live-DB skip pattern
|
||||
in tests/test_house_dedup_merge.py."""
|
||||
dsn = os.environ.get("DATABASE_URL", "")
|
||||
if not dsn or "localhost:5432/test" in dsn:
|
||||
return False
|
||||
try:
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
from app.core.db import engine
|
||||
|
||||
with engine.connect() as conn:
|
||||
conn.execute(_t("SELECT 1"))
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _db_reachable(), reason="no reachable Postgres test DB")
|
||||
def test_real_record_event_inserts_row() -> None:
|
||||
"""End-to-end on the real DB record_event() is bound to (app.core.db.SessionLocal):
|
||||
persists exactly the given fields and is queryable back."""
|
||||
from sqlalchemy import text as _t
|
||||
|
||||
from app.core.db import SessionLocal
|
||||
|
||||
ue.record_event(
|
||||
event_type="pytest_probe",
|
||||
username="pytest_user_events_test",
|
||||
ip="203.0.113.7",
|
||||
user_agent="pytest-real-db",
|
||||
path="/api/v1/trade-in/estimate",
|
||||
method="POST",
|
||||
payload={"probe": True},
|
||||
)
|
||||
with SessionLocal() as db:
|
||||
try:
|
||||
row = db.execute(
|
||||
_t(
|
||||
"SELECT event_type, username, ip_address, user_agent, payload "
|
||||
"FROM user_events WHERE username = :u ORDER BY id DESC LIMIT 1"
|
||||
),
|
||||
{"u": "pytest_user_events_test"},
|
||||
).fetchone()
|
||||
assert row is not None
|
||||
assert row.event_type == "pytest_probe"
|
||||
assert str(row.ip_address) == "203.0.113.7"
|
||||
assert row.payload == {"probe": True}
|
||||
finally:
|
||||
db.execute(
|
||||
_t("DELETE FROM user_events WHERE username = :u"),
|
||||
{"u": "pytest_user_events_test"},
|
||||
)
|
||||
db.commit()
|
||||
Loading…
Add table
Reference in a new issue