Refactor for specs requirements
This commit is contained in:
parent
280dabfd95
commit
539bd8c0f9
@ -60,6 +60,8 @@ def map_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, str, str, str | Non
|
|||||||
if any(low.startswith(prefix) for prefix in VALIDATION_PREFIXES):
|
if any(low.startswith(prefix) for prefix in VALIDATION_PREFIXES):
|
||||||
code, field = _parse_prefixed_validation_error(message)
|
code, field = _parse_prefixed_validation_error(message)
|
||||||
return status.HTTP_422_UNPROCESSABLE_ENTITY, code, message, field
|
return status.HTTP_422_UNPROCESSABLE_ENTITY, code, message, field
|
||||||
|
if low.startswith("role_not_allowed:") or low.startswith("window_closed:"):
|
||||||
|
return status.HTTP_403_FORBIDDEN, "access_denied", message, None
|
||||||
if "не существует" in low or "не принадлежит" in low or "not found" in low:
|
if "не существует" in low or "не принадлежит" in low or "not found" in low:
|
||||||
return status.HTTP_404_NOT_FOUND, "not_found", message, None
|
return status.HTTP_404_NOT_FOUND, "not_found", message, None
|
||||||
if "permission denied" in low or "insufficient privilege" in low:
|
if "permission denied" in low or "insufficient privilege" in low:
|
||||||
@ -68,8 +70,10 @@ def map_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, str, str, str | Non
|
|||||||
if isinstance(exc, IntegrityError):
|
if isinstance(exc, IntegrityError):
|
||||||
if "unique" in low or "duplicate" in low:
|
if "unique" in low or "duplicate" in low:
|
||||||
return status.HTTP_409_CONFLICT, "conflict", message, None
|
return status.HTTP_409_CONFLICT, "conflict", message, None
|
||||||
if "foreign key" in low or "violates check constraint" in low:
|
if "foreign key" in low:
|
||||||
return status.HTTP_409_CONFLICT, "conflict", message, None
|
return status.HTTP_409_CONFLICT, "conflict", message, None
|
||||||
|
if "violates check constraint" in low:
|
||||||
|
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
|
||||||
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
|
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
|
||||||
|
|
||||||
if isinstance(exc, OperationalError):
|
if isinstance(exc, OperationalError):
|
||||||
|
|||||||
@ -23,6 +23,21 @@ def test_map_sqlalchemy_error_unique_conflict():
|
|||||||
assert field is None
|
assert field is None
|
||||||
|
|
||||||
|
|
||||||
|
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 == "db_error"
|
||||||
|
assert "violates check constraint" in message
|
||||||
|
assert field is None
|
||||||
|
|
||||||
|
|
||||||
def test_map_sqlalchemy_error_validation_prefix():
|
def test_map_sqlalchemy_error_validation_prefix():
|
||||||
exc = SQLAlchemyError()
|
exc = SQLAlchemyError()
|
||||||
exc.orig = Exception("unknown_column: q5.plan")
|
exc.orig = Exception("unknown_column: q5.plan")
|
||||||
@ -47,6 +62,24 @@ def test_map_sqlalchemy_error_not_found():
|
|||||||
assert field is None
|
assert field is None
|
||||||
|
|
||||||
|
|
||||||
|
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 == "access_denied"
|
||||||
|
assert message == "role_not_allowed: q1.m1"
|
||||||
|
assert field is None
|
||||||
|
|
||||||
|
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 == "access_denied"
|
||||||
|
assert message.startswith("window_closed:")
|
||||||
|
assert field is None
|
||||||
|
|
||||||
|
|
||||||
def test_map_sqlalchemy_error_operational():
|
def test_map_sqlalchemy_error_operational():
|
||||||
exc = OperationalError(
|
exc = OperationalError(
|
||||||
statement="SELECT 1",
|
statement="SELECT 1",
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user