DFiP_Budget_planing/api/tests/unit/test_exception_handlers.py

from fastapi import HTTPException, status
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError

from src.core.exception_handlers import (
    _error_payload,
    map_http_exception,
    map_sqlalchemy_error,
)


def test_map_sqlalchemy_error_unique_conflict():
    exc = IntegrityError(
        statement="INSERT INTO users ...",
        params={},
        orig=Exception("duplicate key value violates unique constraint users_username_key"),
    )

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_409_CONFLICT
    assert code == status.HTTP_409_CONFLICT
    assert "duplicate key value" in message
    assert field == ""


def test_map_sqlalchemy_error_check_constraint_is_400():
    exc = IntegrityError(
        statement="UPDATE v3.project ...",
        params={},
        orig=Exception("new row for relation violates check constraint chk_v3_project_name"),
    )

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_400_BAD_REQUEST
    assert code == status.HTTP_400_BAD_REQUEST
    assert "violates check constraint" in message
    assert field == ""


def test_map_sqlalchemy_error_validation_prefix():
    exc = SQLAlchemyError()
    exc.orig = Exception("unknown_column: q5.plan")

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert message == "unknown_column: q5.plan"
    assert field == "q5.plan"


def test_map_sqlalchemy_error_validation_prefix_asyncpg_wrapped():
    exc = SQLAlchemyError()
    exc.orig = Exception("<class 'asyncpg.exceptions.RaiseError'>: invalid_formula: chars")

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert message == "invalid_formula: chars"
    assert "asyncpg" not in message
    assert field == "chars"


def test_map_sqlalchemy_error_asyncpg_wrapped_non_validation_keeps_message():
    exc = SQLAlchemyError()
    exc.orig = Exception("<class 'asyncpg.exceptions.RaiseError'>: something else")

    status_code, _, message, _ = map_sqlalchemy_error(exc)

    assert status_code != status.HTTP_422_UNPROCESSABLE_ENTITY


def test_map_sqlalchemy_error_not_found():
    exc = SQLAlchemyError()
    exc.orig = Exception("budget_form #99 не существует")

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_404_NOT_FOUND
    assert code == status.HTTP_404_NOT_FOUND
    assert "не существует" in message
    assert field == ""


def test_map_sqlalchemy_error_role_window_prefixes():
    role_exc = SQLAlchemyError()
    role_exc.orig = Exception("role_not_allowed: q1.m1")
    status_code, code, message, field = map_sqlalchemy_error(role_exc)
    assert status_code == status.HTTP_403_FORBIDDEN
    assert code == status.HTTP_403_FORBIDDEN
    assert message == "role_not_allowed: q1.m1"
    assert field == ""

    window_exc = SQLAlchemyError()
    window_exc.orig = Exception("window_closed: q1.m1 (closed 2026-05-01T00:00:00Z)")
    status_code, code, message, field = map_sqlalchemy_error(window_exc)
    assert status_code == status.HTTP_403_FORBIDDEN
    assert code == status.HTTP_403_FORBIDDEN
    assert message.startswith("window_closed:")
    assert field == ""


def test_map_sqlalchemy_error_operational():
    exc = OperationalError(
        statement="SELECT 1",
        params={},
        orig=Exception("connection refused"),
    )

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_503_SERVICE_UNAVAILABLE
    assert code == status.HTTP_503_SERVICE_UNAVAILABLE
    assert message == "Database unavailable"
    assert field == ""


def test_error_payload_matches_spec_shape():
    payload = _error_payload(
        status.HTTP_422_UNPROCESSABLE_ENTITY,
        "unknown_column: q5.plan",
        "q5.plan",
    )

    assert payload == {
        "code": status.HTTP_422_UNPROCESSABLE_ENTITY,
        "field": "q5.plan",
        "message": "unknown_column: q5.plan",
    }


def test_map_http_exception_validation_prefix():
    exc = HTTPException(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        detail="unknown_column: q5.plan",
    )

    status_code, code, message, field = map_http_exception(exc)

    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert message == "unknown_column: q5.plan"
    assert field == ""


def test_map_http_exception_from_structured_detail():
    exc = HTTPException(
        status_code=status.HTTP_400_BAD_REQUEST,
        detail={"code": "validation_error", "message": "bad input", "field": "username"},
    )

    status_code, code, message, field = map_http_exception(exc)

    assert status_code == status.HTTP_400_BAD_REQUEST
    assert code == status.HTTP_400_BAD_REQUEST
    assert message == "bad input"
    assert field == "username"


def test_map_http_exception_unauthorized():
    exc = HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Недействительный токен")

    status_code, code, message, field = map_http_exception(exc)

    assert status_code == status.HTTP_401_UNAUTHORIZED
    assert code == status.HTTP_401_UNAUTHORIZED
    assert message == "Недействительный токен"
    assert field == ""


def test_map_http_exception_not_found_maps_code_from_status():
    exc = HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="not found")

    status_code, code, message, field = map_http_exception(exc)

    assert status_code == status.HTTP_404_NOT_FOUND
    assert code == status.HTTP_404_NOT_FOUND
    assert message == "not found"
    assert field == ""


def test_map_sqlalchemy_error_invalid_formula_prefix():
    exc = SQLAlchemyError()
    exc.orig = Exception("invalid_formula: expression")

    status_code, code, message, field = map_sqlalchemy_error(exc)

    assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
    assert message == "invalid_formula: expression"
    assert field == "expression"