160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
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_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 == ""
|