112 lines
3.3 KiB
Python
112 lines
3.3 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 == "conflict"
|
|
assert "duplicate key value" in message
|
|
assert field is None
|
|
|
|
|
|
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 == "unknown_column"
|
|
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 == "not_found"
|
|
assert "не существует" in message
|
|
assert field is None
|
|
|
|
|
|
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 == "db_unavailable"
|
|
assert message == "Database unavailable"
|
|
assert field is None
|
|
|
|
|
|
def test_error_payload_matches_spec_shape():
|
|
payload = _error_payload("unknown_column", "unknown_column: q5.plan", "q5.plan")
|
|
|
|
assert payload == {
|
|
"code": "unknown_column",
|
|
"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 == "unknown_column"
|
|
assert message == "unknown_column: q5.plan"
|
|
assert field == "q5.plan"
|
|
|
|
|
|
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 == "validation_error"
|
|
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 == "unauthorized"
|
|
assert message == "Недействительный токен"
|
|
assert field is None
|