arm-core/api/tests/test_exception_handlers.py

"""Обработчики ошибок БД (`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 import UniqueConstraint
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError

from src.common.constraint_conflicts import UNIQUE_CONSTRAINT_CONFLICTS
from src.core.exception_handlers import LOG_STATEMENT_LIMIT, register_exception_handlers
from src.models.base import Base

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


# --- гонки уникальных ограничений: реестр `common/constraint_conflicts.py` -------------------------------------------


class _AsyncpgCause(Exception):
    """Как `asyncpg.UniqueViolationError`: структурированное имя ограничения; текст — с DETAIL и значением."""

    def __init__(self, constraint_name: str | None, sqlstate: str = "23505") -> None:
        super().__init__(f'duplicate key value violates unique constraint "x"\nDETAIL:  Key (email)=({SECRET}) exists.')
        self.constraint_name = constraint_name
        self.sqlstate = sqlstate


class _AdaptedDriverError(Exception):
    """Как `AsyncAdapt_asyncpg_dbapi.IntegrityError`: `sqlstate` на адаптере, `__cause__` — исключение asyncpg."""

    def __init__(self, cause: _AsyncpgCause) -> None:
        super().__init__(str(cause))
        self.sqlstate = cause.sqlstate
        self.__cause__ = cause


class _Diag:
    def __init__(self, constraint_name: str) -> None:
        self.constraint_name = constraint_name


class _PsycopgError(Exception):
    """Как `psycopg.errors.UniqueViolation`: `sqlstate` и `diag.constraint_name` на самом исключении."""

    sqlstate = "23505"

    def __init__(self, constraint_name: str) -> None:
        super().__init__(f"duplicate key: {SECRET}")
        self.diag = _Diag(constraint_name)


def _integrity(orig: Exception) -> IntegrityError:
    return IntegrityError("INSERT INTO dkb.app_user (email) VALUES ($1)", (SECRET,), orig)


def _app_raising(exc: Exception) -> FastAPI:
    app = FastAPI()
    register_exception_handlers(app)

    @app.get("/boom")
    async def boom() -> None:
        raise exc

    return app


async def _get_raising(exc: Exception) -> httpx.Response:
    transport = httpx.ASGITransport(app=_app_raising(exc))
    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
        return await client.get("/boom")


def test_registry_keys_exist_as_unique_constraints_or_indexes_in_metadata() -> None:
    """Переименование ограничения в ORM ломает реестр молча (гонка снова отдаёт общий 409) — ловим здесь."""
    unique_names: set[str] = set()
    for table in Base.metadata.tables.values():
        unique_names |= {c.name for c in table.constraints if isinstance(c, UniqueConstraint) and c.name}
        unique_names |= {i.name for i in table.indexes if i.unique and i.name}
    assert set(UNIQUE_CONSTRAINT_CONFLICTS) <= unique_names


def test_registry_codes_and_messages_match_service_precheck_texts() -> None:
    codes = {name: conflict.code for name, conflict in UNIQUE_CONSTRAINT_CONFLICTS.items()}
    assert codes == {
        "uq_company_group_crm_id": "CRM_ID_EXISTS",
        "uq_legal_entity_crm_id": "CRM_ID_EXISTS",
        "uq_related_group_crm_id": "CRM_ID_EXISTS",
        "uq_legal_entity_inn": "INN_EXISTS",
        "uq_legal_entity_alias_legal_entity_id_normalized_alias": "ALIAS_EXISTS",
        "uq_app_user_email": "EMAIL_EXISTS",
        "uq_related_group_member_group_entity_current": "MEMBER_EXISTS",
        "uq_related_group_member_entity_kind_current": "MEMBER_OF_OTHER_GROUP",
        "uq_consolidation_group_year_quarter": "CONSOLIDATION_EXISTS",
        "uq_consolidation_lock_version_active": "LOCKED_BY_OTHER",
    }
    # у pre-check-ов постоянные тексты — тот же `message`, динамических значений в реестре нет
    assert UNIQUE_CONSTRAINT_CONFLICTS["uq_legal_entity_inn"].message == "ЮЛ с таким ИНН уже существует"
    assert UNIQUE_CONSTRAINT_CONFLICTS["uq_app_user_email"].message == "Пользователь с таким email уже существует"


@pytest.mark.parametrize(
    "orig",
    [
        _AdaptedDriverError(_AsyncpgCause("uq_app_user_email")),
        _PsycopgError("uq_app_user_email"),
    ],
    ids=["asyncpg-cause", "psycopg-diag"],
)
async def test_known_unique_violation_gives_coded_conflict_and_logs_code(records: list[Any], orig: Exception) -> None:
    response = await _get_raising(_integrity(orig))

    assert response.status_code == HTTPStatus.CONFLICT
    assert response.json() == {
        "detail": {"code": "EMAIL_EXISTS", "message": "Пользователь с таким email уже существует"}
    }
    assert SECRET not in response.text
    (record,) = [r for r in records if r["level"].name == "ERROR"]
    assert "resolved_code=EMAIL_EXISTS" in record["message"]
    assert SECRET not in record["message"]
    assert "uq_app_user_email" not in response.text


@pytest.mark.parametrize(
    "orig",
    [
        _AdaptedDriverError(_AsyncpgCause("uq_unknown_constraint")),  # unique, но не из реестра
        _AdaptedDriverError(_AsyncpgCause(None)),  # имя не определилось
        _AdaptedDriverError(_AsyncpgCause("uq_app_user_email", sqlstate="23503")),  # FK, не unique
        _AdaptedDriverError(_AsyncpgCause("uq_app_user_email", sqlstate="23514")),  # check violation
        _DriverError("no constraint name"),
    ],
    ids=["unknown-unique", "no-name", "foreign-key", "check", "bare-driver-error"],
)
async def test_other_integrity_errors_stay_generic(records: list[Any], orig: Exception) -> None:
    response = await _get_raising(_integrity(orig))

    assert response.status_code == HTTPStatus.CONFLICT
    assert response.json() == {"detail": "нарушение целостности данных в БД"}
    (record,) = [r for r in records if r["level"].name == "ERROR"]
    assert "resolved_code=None" in record["message"]
    assert SECRET not in record["message"]