6a6ffb54f4
Обработчики IntegrityError (409) и SQLAlchemyError (500) молча отдавали ответ. Теперь каждая такая ошибка — одна запись ERROR: путь и метод запроса, классы ошибок SQLAlchemy и драйвера, SQLSTATE, SQL-текст с плейсхолдерами (усечён до 200 символов) и стек вызовов. Текст исключения в лог не идёт: str() ошибки asyncpg/SQLAlchemy содержит параметры запроса и DETAIL PostgreSQL со значениями ключей (email, ИНН). Ответы прежние. JwksVerifier.verify: TypeError/ValueError, превращаемые в AuthError, теперь пишут warning в том же санированном виде repr(str(exc))[:200]. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
107 lines
4.1 KiB
Python
107 lines
4.1 KiB
Python
"""Обработчики ошибок БД (`src/core/exception_handlers.py`, финальное ревью I-2): ответ прежний, в лог — одна запись
|
||
ERROR без параметров запроса и DETAIL PostgreSQL (персональные данные)."""
|
||
|
||
from collections.abc import Iterator
|
||
from http import HTTPStatus
|
||
from typing import Any
|
||
|
||
import httpx
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from loguru import logger
|
||
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError
|
||
|
||
from src.core.exception_handlers import LOG_STATEMENT_LIMIT, register_exception_handlers
|
||
|
||
SECRET = "ivan.petrov@rshb.ru"
|
||
|
||
|
||
class _DriverError(Exception):
|
||
"""Ошибка драйвера: как у asyncpg, текст содержит DETAIL со значением ключа, есть `sqlstate`."""
|
||
|
||
sqlstate = "23505"
|
||
|
||
|
||
def _app() -> FastAPI:
|
||
app = FastAPI()
|
||
register_exception_handlers(app)
|
||
|
||
@app.get("/integrity")
|
||
async def integrity() -> None:
|
||
raise IntegrityError(
|
||
"INSERT INTO dkb.app_user (email) VALUES ($1)",
|
||
(SECRET,),
|
||
_DriverError(f"duplicate key\nDETAIL: Key (email)=({SECRET}) already exists."),
|
||
)
|
||
|
||
@app.get("/operational")
|
||
async def operational() -> None:
|
||
raise OperationalError("SELECT " + "x" * 1000 + " WHERE email = $1", (SECRET,), ConnectionError(SECRET))
|
||
|
||
@app.get("/generic")
|
||
async def generic() -> None:
|
||
raise SQLAlchemyError(f"ошибка без DBAPI: {SECRET}")
|
||
|
||
return app
|
||
|
||
|
||
@pytest.fixture
|
||
def records() -> Iterator[list[Any]]:
|
||
"""Записи loguru уровня DEBUG и выше за время теста (временный sink)."""
|
||
collected: list[Any] = []
|
||
handler_id = logger.add(lambda message: collected.append(message.record), level="DEBUG")
|
||
try:
|
||
yield collected
|
||
finally:
|
||
logger.remove(handler_id)
|
||
|
||
|
||
async def _get(path: str) -> httpx.Response:
|
||
transport = httpx.ASGITransport(app=_app())
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
|
||
return await client.get(path)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("path", "status", "detail", "error_type"),
|
||
[
|
||
("/integrity", HTTPStatus.CONFLICT, "нарушение целостности данных в БД", "IntegrityError"),
|
||
("/operational", HTTPStatus.INTERNAL_SERVER_ERROR, "ошибка базы данных", "OperationalError"),
|
||
("/generic", HTTPStatus.INTERNAL_SERVER_ERROR, "ошибка базы данных", "SQLAlchemyError"),
|
||
],
|
||
)
|
||
async def test_db_error_logged_once_without_pii(
|
||
records: list[Any], path: str, status: HTTPStatus, detail: str, error_type: str
|
||
) -> None:
|
||
response = await _get(path)
|
||
|
||
assert response.status_code == status
|
||
assert response.json() == {"detail": detail}
|
||
errors = [r for r in records if r["level"].no >= logger.level("ERROR").no]
|
||
assert len(errors) == 1
|
||
(record,) = errors
|
||
assert record["level"].name == "ERROR"
|
||
message = record["message"]
|
||
assert f"path={path}" in message
|
||
assert "method=GET" in message
|
||
assert f"error_type={error_type}" in message
|
||
assert record["exception"] is None # текст исключения (параметры, DETAIL) в лог не попадает
|
||
assert SECRET not in message
|
||
assert "test_exception_handlers.py" in message # стек вызовов — есть
|
||
|
||
|
||
async def test_integrity_error_log_has_driver_class_sqlstate_and_statement(records: list[Any]) -> None:
|
||
await _get("/integrity")
|
||
(record,) = [r for r in records if r["level"].name == "ERROR"]
|
||
assert "db_error_type=_DriverError" in record["message"]
|
||
assert "sqlstate=23505" in record["message"]
|
||
assert "statement='INSERT INTO dkb.app_user (email) VALUES ($1)'" in record["message"]
|
||
|
||
|
||
async def test_long_statement_is_truncated(records: list[Any]) -> None:
|
||
await _get("/operational")
|
||
(record,) = [r for r in records if r["level"].name == "ERROR"]
|
||
statement = record["extra"]["statement"]
|
||
assert len(statement) <= LOG_STATEMENT_LIMIT + 2 # repr добавляет кавычки
|
||
assert "WHERE email" not in statement
|