RSHB/DFiP_Budget_planing
5
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'test' into front-logs

Browse Source
This commit is contained in:
PotapovaA 2026-06-17 11:22:56 +03:00
commit 8a02913f2e
11 changed files with 3296 additions and 168 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

96
api/src/core/exception_handlers.py
View File

@ -1,4 +1,5 @@
from fastapi import FastAPI, HTTPException, status
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse
from sqlalchemy.exc import IntegrityError, OperationalError, SQLAlchemyError, DBAPIError
@ -18,7 +19,7 @@ from src.core.CONSTANTS import (
PG_403_ERRORS,
)
def _error_payload(code: str | int, message: str, field: str | None = None) -> dict:
def _error_payload(code: int, message: str, field: str = "") -> dict:
"""Формирует структурированный payload ошибки по спеке."""
return {
"code": code,
@ -27,13 +28,12 @@ def _error_payload(code: str | int, message: str, field: str | None = None) -> d
}
def _parse_prefixed_validation_error(message: str) -> tuple[str, str | None]:
def _parse_prefixed_validation_error(message: str) -> str:
for prefix in VALIDATION_PREFIXES:
if message.lower().startswith(prefix):
code = prefix[:-1]
field = message.split(":", 1)[1].strip() if ":" in message else None
return code, field or None
return "validation_error", None
field = message.split(":", 1)[1].strip() if ":" in message else ""
return field or ""
return ""
def _first_db_message_line(message: str) -> str:
@ -45,34 +45,42 @@ def _first_db_message_line(message: str) -> str:
return message.strip()
def map_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, str, str, str | None]:
def map_sqlalchemy_error(exc: SQLAlchemyError) -> tuple[int, int, str, str]:
"""Маппинг ошибок SQLAlchemy в HTTP-статус + code/message/field."""
raw = str(getattr(exc, "orig", exc) or exc).strip()
message = _first_db_message_line(raw) or "Database error"
low = message.lower()
if any(low.startswith(prefix) for prefix in VALIDATION_PREFIXES):
code, field = _parse_prefixed_validation_error(message)
return status.HTTP_422_UNPROCESSABLE_ENTITY, code, message, field
status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
field = _parse_prefixed_validation_error(message)
return status_code, status_code, message, field
if any(error in low for error in PG_404_ERRORS):
return status.HTTP_404_NOT_FOUND, "not_found", message, None
status_code = status.HTTP_404_NOT_FOUND
return status_code, status_code, message, ""
if any(error in low for error in PG_403_ERRORS):
return status.HTTP_403_FORBIDDEN, "access_denied", message, None
status_code = status.HTTP_403_FORBIDDEN
return status_code, status_code, message, ""
if isinstance(exc, IntegrityError):
if any(error in low for error in PG_409_ERRORS):
return status.HTTP_409_CONFLICT, "conflict", message, None
status_code = status.HTTP_409_CONFLICT
return status_code, status_code, message, ""
if any(error in low for error in PG_400_ERRORS):
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
status_code = status.HTTP_400_BAD_REQUEST
return status_code, status_code, message, ""
status_code = status.HTTP_400_BAD_REQUEST
return status_code, status_code, message, ""
if isinstance(exc, OperationalError):
return status.HTTP_503_SERVICE_UNAVAILABLE, "db_unavailable", "Database unavailable", None
status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return status_code, status_code, "Database unavailable", ""
return status.HTTP_400_BAD_REQUEST, "db_error", message, None
status_code = status.HTTP_400_BAD_REQUEST
return status_code, status_code, message, ""
def map_http_exception(exc: HTTPException) -> tuple[int, str | int, str, str | None]:
def map_http_exception(exc: HTTPException) -> tuple[int, int, str, str]:
"""Маппинг HTTPException в единый формат ошибки API."""
status_code = int(exc.status_code)
detail = exc.detail
@ -80,16 +88,35 @@ def map_http_exception(exc: HTTPException) -> tuple[int, str | int, str, str | N
if isinstance(detail, dict):
message = str(detail.get("message") or detail.get("detail") or "HTTP error")
field = detail.get("field")
if field is not None:
field = str(field)
code = detail.get("code", status_code)
return status_code, code, message, field
field_str = str(field) if field is not None else ""
return status_code, status_code, message, field_str
message = str(detail) if detail else "HTTP error"
return status_code, status_code, message, None
return status_code, status_code, message, ""
def map_request_validation_error(exc: RequestValidationError) -> tuple[int, int, str, str]:
status_code = status.HTTP_422_UNPROCESSABLE_ENTITY
errors = exc.errors()
if not errors:
return status_code, status_code, "Validation error", ""
first_error = errors[0]
message = str(first_error.get("msg") or "Validation error")
loc = first_error.get("loc") or ()
field = ".".join(str(part) for part in loc if part not in ("body", "query", "path", "header", "cookie"))
return status_code, status_code, message, field
def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(RequestValidationError)
async def request_validation_exception_handler(request, exc: RequestValidationError):
status_code, code, message, field = map_request_validation_error(exc)
return JSONResponse(
status_code=status_code,
content=_error_payload(code, message, field=field),
)
@app.exception_handler(HTTPException)
async def http_exception_handler(request, exc: HTTPException):
status_code, code, message, field = map_http_exception(exc)
@ -101,38 +128,43 @@ def register_exception_handlers(app: FastAPI) -> None:
@app.exception_handler(AccessDeniedException)
async def access_denied_exception_handler(request, exc: AccessDeniedException):
status_code = status.HTTP_403_FORBIDDEN
return JSONResponse(
status_code=status.HTTP_403_FORBIDDEN,
content=_error_payload("access_denied", exc.description),
status_code=status_code,
content=_error_payload(status_code, exc.description),
)
@app.exception_handler(ValidationException)
async def validation_exception_handler(request, exc: ValidationException):
status_code = status.HTTP_400_BAD_REQUEST
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content=_error_payload("validation_error", exc.description, field=exc.field),
status_code=status_code,
content=_error_payload(status_code, exc.description, field=exc.field or ""),
)
@app.exception_handler(UserNotFoundException)
async def user_not_found_exception_handler(request, exc: UserNotFoundException):
status_code = status.HTTP_404_NOT_FOUND
return JSONResponse(
status_code=status.HTTP_404_NOT_FOUND,
content=_error_payload("user_not_found", exc.description),
status_code=status_code,
content=_error_payload(status_code, exc.description),
)
@app.exception_handler(UsernameConflictException)
async def username_conflict_exception_handler(request, exc: UsernameConflictException):
status_code = status.HTTP_409_CONFLICT
return JSONResponse(
status_code=status.HTTP_409_CONFLICT,
content=_error_payload("username_conflict", exc.description),
status_code=status_code,
content=_error_payload(status_code, exc.description),
)
@app.exception_handler(BasicAppException)
async def basic_app_exception_handler(request, exc: BasicAppException):
message = exc.description or "Application error"
status_code = status.HTTP_400_BAD_REQUEST
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content=_error_payload("application_error", message),
status_code=status_code,
content=_error_payload(status_code, message),
)
@app.exception_handler(SQLAlchemyError)

1
api/src/services/export_service/__init__.py Normal file
View File

@ -0,0 +1 @@
from .export_service import ExportService

2912
api/src/services/export_service/export_mappings.py Normal file
View File

File diff suppressed because it is too large Load Diff

35
api/src/services/export_normalizers.py → api/src/services/export_service/export_normalizers.py
View File

@ -194,6 +194,41 @@ class Form1RowNormalizer:
return None
class Form1DepthRowNormalizer:
def __init__(self, serializer: ExcelValueSerializer):
self.serializer = serializer
def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
normalized = dict()
self._merge_header_fields_into_row(normalized=normalized, original=data)
self._normilize_depth(normalized=normalized, original=data)
return normalized
def _merge_header_fields_into_row(self, normalized: dict[str, Any], original: dict[str, Any]) -> None:
header = original.get("header")
if isinstance(header, dict):
if "section" in header and "section" not in normalized:
normalized["section"] = header.get("section")
if "item_id" in header and "item_id" not in normalized:
normalized["item_id"] = header.get("item_id")
if "name" in header and "name" not in normalized:
normalized["name"] = header.get("name")
if "num_group" in header and "num_group" not in normalized:
normalized["num_group"] = header.get("num_group")
if "internal_order" in header and "internal_order" not in normalized:
normalized["internal_order"] = header.get("internal_order")
normalized.pop("header", None)
normalized.pop("line_id", None)
def _normilize_depth(self, normalized: dict, original: dict, prefix: str = "") -> None:
for key, value in original.items():
if isinstance(value, dict):
self._normilize_depth(normalized=normalized, original=value, prefix=f"{prefix}{key}.")
else:
normalized[f"{prefix}{key}"] = value
class SmetaRowNormalizer:
def normalize(self, data: dict[str, Any]) -> dict[str, Any]:
return {

36
api/src/services/export_service.py → api/src/services/export_service/export_service.py
View File

@ -15,16 +15,20 @@ from sqlalchemy.ext.asyncio import AsyncSession
from src.db.models.app_user import AppUser
from src.db.models.form_type import FormTypeEnum
from src.services.budget_form_service import BudgetFormService
from src.services.export_normalizers import (
from src.services.export_service.export_normalizers import (
ExcelValueSerializer,
Form1RowNormalizer,
Form1DepthRowNormalizer,
Form3ReportRowNormalizer,
SmetaRowNormalizer,
)
from src.services.export_writers import (
FORM1_FLAT_COLUMNS,
from src.services.export_service.export_mappings import (
FORM1_COLORS,
FORM1_DEPTH_COLUMNS,
FORM3_REPORT_COLUMNS,
SMETA_COLUMNS,
)
from src.services.export_service.export_writers import (
DepthSheetWriter,
TabularSheetWriter,
)
from src.services.project_service import ProjectService
@ -51,12 +55,12 @@ class ExportService:
self.sheet_service = SheetService(db)
self.project_service = ProjectService(db)
self.value_serializer = ExcelValueSerializer()
self.form1_row_normalizer = Form1RowNormalizer(self.value_serializer)
self.form1_row_normalizer = Form1DepthRowNormalizer(self.value_serializer)
self.form3_report_row_normalizer = Form3ReportRowNormalizer()
self.smeta_row_normalizer = SmetaRowNormalizer()
self.sheet_writers: dict[str, TabularSheetWriter] = {
"__default__": TabularSheetWriter(
columns=FORM1_FLAT_COLUMNS,
"__default__": DepthSheetWriter(
columns_dict=FORM1_DEPTH_COLUMNS,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
),
@ -65,6 +69,24 @@ class ExportService:
row_normalizer=self.smeta_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
),
"OPER": DepthSheetWriter(
columns_dict=FORM1_DEPTH_COLUMNS,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
colors_config=FORM1_COLORS["OPER"],
),
"AHR": DepthSheetWriter(
columns_dict=FORM1_DEPTH_COLUMNS,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
colors_config=FORM1_COLORS["AHR"],
),
"CAP": DepthSheetWriter(
columns_dict=FORM1_DEPTH_COLUMNS,
row_normalizer=self.form1_row_normalizer.normalize,
value_serializer=self.value_serializer.to_excel_value,
colors_config=FORM1_COLORS["CAP"],
),
}
self.form3_report_writer = TabularSheetWriter(
columns=FORM3_REPORT_COLUMNS,

209
api/src/services/export_service/export_writers.py Normal file
View File

@ -0,0 +1,209 @@
from __future__ import annotations
from collections import defaultdict
from typing import Any, Callable
from openpyxl.styles import Alignment, Border, Color, Font, PatternFill, Side
from openpyxl.worksheet.worksheet import Worksheet
from openpyxl.utils import get_column_letter
from src.services.export_service.export_mappings import PALETTE
RowNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
ValueSerializer = Callable[[Any], Any]
class TabularSheetWriter:
def __init__(
self,
columns: list[tuple[str, str]],
row_normalizer: RowNormalizer,
value_serializer: ValueSerializer,
):
self.columns = columns
self.row_normalizer = row_normalizer
self.value_serializer = value_serializer
def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
worksheet.append([title for _, title in self.columns])
for row in rows:
raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
normalized = self.row_normalizer(raw_data)
worksheet.append(
[self.value_serializer(normalized.get(key)) for key, _ in self.columns]
)
class DepthSheetWriter:
alignment = Alignment(horizontal='center', vertical='center')
thin_border = Border(
left=Side(style='thin'),
right=Side(style='thin'),
top=Side(style='thin'),
bottom=Side(style='thin')
)
def __init__(
self,
columns_dict: dict,
row_normalizer: RowNormalizer,
value_serializer: ValueSerializer,
colors_config: dict | None = None,
):
self.columns_dict = columns_dict
self._flat_columns_to_depth = {}
self.columns = self._columns_dict_to_flat(columns_dict)
self.row_normalizer = row_normalizer
self.value_serializer = value_serializer
self.colors_config = colors_config
self.palette = None
if colors_config:
self._set_palette()
def _columns_dict_to_flat(
self,
columns_dict: dict | None = None,
prefix: str = "",
depth_prefix: str = "",
) -> None:
if columns_dict is None:
columns_dict = self.columns_dict
result = []
for key, sub_dict in columns_dict.items():
if "key" in sub_dict:
new_key = sub_dict['key']
else:
new_key = key
if "children" not in sub_dict:
if new_key:
add_key = f"{prefix}{new_key}"
result.append(add_key)
self._flat_columns_to_depth[add_key] = f"{depth_prefix}{key}"
else:
result += self._columns_dict_to_flat(
sub_dict["children"],
prefix=f"{prefix}{new_key}." if new_key else prefix,
depth_prefix=f"{depth_prefix}{key}.",
)
return result
def _set_palette(self) -> None:
self.palette = {}
for key in self.columns:
full_key = None
for sub_key in self._flat_columns_to_depth[key].split("."):
if full_key is None:
full_key = sub_key
else:
full_key = f"{full_key}.{sub_key}"
if full_key in self.colors_config and "palette" in self.colors_config[full_key]:
self.palette[key] = self.colors_config[full_key]["palette"]
def _get_header_max_row(self, column_dict: dict) -> int:
max_depth = 0
for sub_dict in column_dict.values():
if "children" in sub_dict:
max_depth = max(max_depth, self._get_header_max_row(sub_dict["children"]))
return max_depth + 1
def _write_header(
self,
worksheet: Worksheet,
columns_dict: dict | None = None,
current_column: int = 1,
current_row: int = 1,
max_row: int | None = None,
full_key: str | None = None
) -> int:
if columns_dict is None:
columns_dict = self.columns_dict
max_row = self._get_header_max_row(self.columns_dict)
for key, sub_dict in columns_dict.items():
if full_key:
key = f"{full_key}.{key}"
if "children" not in sub_dict:
cell = worksheet.cell(row=current_row, column=current_column, value=sub_dict["name"])
cell.border = self.thin_border
if max_row != current_row:
worksheet.merge_cells(
start_row=current_row,
start_column=current_column,
end_row=max_row,
end_column=current_column,
)
current_column += 1
else:
new_column = self._write_header(
worksheet=worksheet,
columns_dict=sub_dict["children"],
current_column=current_column,
current_row=current_row+1,
max_row=max_row,
full_key=key,
)
cell = worksheet.cell(row=current_row, column=current_column, value=sub_dict["name"])
if new_column - current_column > 1:
worksheet.merge_cells(
start_row=current_row,
start_column=current_column,
end_row=current_row,
end_column=new_column - 1,
)
cell.border = self.thin_border
current_column = new_column
cell.alignment = self.alignment
if self.colors_config and (color := self.colors_config.get(key)):
if "background" in color:
cell.fill = PatternFill(
start_color=Color(rgb=color["background"]),
fill_type="solid",
)
if "color" in color:
cell.font = Font(color=color["color"])
return current_column
def _set_widths(self, worksheet: Worksheet):
for col in worksheet.columns:
max_len = 0
col_letter = get_column_letter(col[0].column) # Get alphabetical column name (e.g., 'A')
for cell in col:
if cell.value is not None:
cell_len = len(str(cell.value))
if cell_len > max_len:
max_len = cell_len
adjusted_width = max(max_len + 3, 10)
worksheet.column_dimensions[col_letter].width = adjusted_width
def _set_color_to_row(self, ws: Worksheet, item_type: str) -> None:
if not self.palette:
return
row = ws.max_row
for i, key in enumerate(self.columns):
if key not in self.palette:
continue
cell = ws.cell(column=i+1, row=row)
if (palette := self.palette.get(key)):
cell.fill = PatternFill(
start_color=Color(rgb=PALETTE[palette][item_type]),
fill_type="solid",
)
def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
self._write_header(worksheet)
for row in rows:
raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
normalized = self.row_normalizer(raw_data)
worksheet.append(
[self.value_serializer(normalized.get(key)) for key in self.columns]
)
if len(row) > 3:
self._set_color_to_row(ws=worksheet, item_type=row[0])
self._set_widths(worksheet)

103
api/src/services/export_writers.py
View File

@ -1,103 +0,0 @@
from __future__ import annotations
from typing import Any, Callable
from openpyxl.worksheet.worksheet import Worksheet
FORM1_FLAT_COLUMNS: list[tuple[str, str]] = [
("section", "Код раздела"),
("item_id", "ID статьи"),
("name", "Наименование"),
("q1", "I квартал"),
("q2", "II квартал"),
("q3", "III квартал"),
("q4", "IV квартал"),
("totals", "ИТОГО"),
("plan", "План"),
("reserve", "Резерв"),
("approved", "Согласовано"),
("collegial", "Коллегиальный"),
("allocation", "Аллокация"),
("sequestration", "Секвестр"),
("contract", "Договор"),
("contract_sum", "Сумма договора"),
("ckk", "ЦКК"),
]
SMETA_COLUMNS: list[tuple[str, str]] = [
("section_code", "Код раздела"),
("name", "Наименование"),
("plan_support_q1", "План поддержка I квартал"),
("plan_support_q2", "План поддержка II квартал"),
("plan_support_q3", "План поддержка III квартал"),
("plan_support_q4", "План поддержка IV квартал"),
("plan_support_year", "План поддержка Год"),
("plan_development_q1", "План развитие I квартал"),
("plan_development_q2", "План развитие II квартал"),
("plan_development_q3", "План развитие III квартал"),
("plan_development_q4", "План развитие IV квартал"),
("plan_development_year", "План развитие Год"),
("plan_total_year", "План ИТОГО Год"),
("approved_support_q1", "Утверждено поддержка I квартал"),
("approved_support_q2", "Утверждено поддержка II квартал"),
("approved_support_q3", "Утверждено поддержка III квартал"),
("approved_support_q4", "Утверждено поддержка IV квартал"),
("approved_support_year", "Утверждено поддержка Год"),
("approved_development_q1", "Утверждено развитие I квартал"),
("approved_development_q2", "Утверждено развитие II квартал"),
("approved_development_q3", "Утверждено развитие III квартал"),
("approved_development_q4", "Утверждено развитие IV квартал"),
("approved_development_year", "Утверждено развитие Год"),
("approved_total_year", "Утверждено ИТОГО Год"),
("fact_support_q1", "Факт поддержка I квартал"),
("fact_support_q2", "Факт поддержка II квартал"),
("fact_support_q3", "Факт поддержка III квартал"),
("fact_support_q4", "Факт поддержка IV квартал"),
("fact_support_year", "Факт поддержка Год"),
("fact_development_q1", "Факт развитие I квартал"),
("fact_development_q2", "Факт развитие II квартал"),
("fact_development_q3", "Факт развитие III квартал"),
("fact_development_q4", "Факт развитие IV квартал"),
("fact_development_year", "Факт развитие Год"),
("fact_total_year", "Факт ИТОГО Год"),
("corrected_support_q2", "Корректировка поддержка II квартал"),
("corrected_support_q3", "Корректировка поддержка III квартал"),
("corrected_support_q4", "Корректировка поддержка IV квартал"),
("corrected_development_q2", "Корректировка развитие II квартал"),
("corrected_development_q3", "Корректировка развитие III квартал"),
("corrected_development_q4", "Корректировка развитие IV квартал"),
]
FORM3_REPORT_COLUMNS: list[tuple[str, str]] = [
("section", "Код раздела"),
("item_id", "ID статьи"),
("name", "Наименование"),
("q1", "I квартал"),
("q2", "II квартал"),
("q3", "III квартал"),
("q4", "IV квартал"),
("totals", "ИТОГО"),
]
RowNormalizer = Callable[[dict[str, Any]], dict[str, Any]]
ValueSerializer = Callable[[Any], Any]
class TabularSheetWriter:
def __init__(
self,
columns: list[tuple[str, str]],
row_normalizer: RowNormalizer,
value_serializer: ValueSerializer,
):
self.columns = columns
self.row_normalizer = row_normalizer
self.value_serializer = value_serializer
def write(self, worksheet: Worksheet, rows: list[tuple]) -> None:
worksheet.append([title for _, title in self.columns])
for row in rows:
raw_data = row[3] if len(row) > 3 and isinstance(row[3], dict) else {}
normalized = self.row_normalizer(raw_data)
worksheet.append(
[self.value_serializer(normalized.get(key)) for key, _ in self.columns]
)

42
api/tests/unit/test_exception_handlers.py
View File

@ -18,9 +18,9 @@ def test_map_sqlalchemy_error_unique_conflict():
status_code, code, message, field = map_sqlalchemy_error(exc)
assert status_code == status.HTTP_409_CONFLICT
assert code == "conflict"
assert code == status.HTTP_409_CONFLICT
assert "duplicate key value" in message
assert field is None
assert field == ""
def test_map_sqlalchemy_error_check_constraint_is_400():
@ -33,9 +33,9 @@ def test_map_sqlalchemy_error_check_constraint_is_400():
status_code, code, message, field = map_sqlalchemy_error(exc)
assert status_code == status.HTTP_400_BAD_REQUEST
assert code == "db_error"
assert code == status.HTTP_400_BAD_REQUEST
assert "violates check constraint" in message
assert field is None
assert field == ""
def test_map_sqlalchemy_error_validation_prefix():
@ -45,7 +45,7 @@ def test_map_sqlalchemy_error_validation_prefix():
status_code, code, message, field = map_sqlalchemy_error(exc)
assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert code == "unknown_column"
assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert message == "unknown_column: q5.plan"
assert field == "q5.plan"
@ -57,9 +57,9 @@ def test_map_sqlalchemy_error_not_found():
status_code, code, message, field = map_sqlalchemy_error(exc)
assert status_code == status.HTTP_404_NOT_FOUND
assert code == "not_found"
assert code == status.HTTP_404_NOT_FOUND
assert "не существует" in message
assert field is None
assert field == ""
def test_map_sqlalchemy_error_role_window_prefixes():
@ -67,17 +67,17 @@ def test_map_sqlalchemy_error_role_window_prefixes():
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 code == status.HTTP_403_FORBIDDEN
assert message == "role_not_allowed: q1.m1"
assert field is None
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 == "access_denied"
assert code == status.HTTP_403_FORBIDDEN
assert message.startswith("window_closed:")
assert field is None
assert field == ""
def test_map_sqlalchemy_error_operational():
@ -90,16 +90,20 @@ def test_map_sqlalchemy_error_operational():
status_code, code, message, field = map_sqlalchemy_error(exc)
assert status_code == status.HTTP_503_SERVICE_UNAVAILABLE
assert code == "db_unavailable"
assert code == status.HTTP_503_SERVICE_UNAVAILABLE
assert message == "Database unavailable"
assert field is None
assert field == ""
def test_error_payload_matches_spec_shape():
payload = _error_payload("unknown_column", "unknown_column: q5.plan", "q5.plan")
payload = _error_payload(
status.HTTP_422_UNPROCESSABLE_ENTITY,
"unknown_column: q5.plan",
"q5.plan",
)
assert payload == {
"code": "unknown_column",
"code": status.HTTP_422_UNPROCESSABLE_ENTITY,
"field": "q5.plan",
"message": "unknown_column: q5.plan",
}
@ -116,7 +120,7 @@ def test_map_http_exception_validation_prefix():
assert status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert message == "unknown_column: q5.plan"
assert field is None
assert field == ""
def test_map_http_exception_from_structured_detail():
@ -128,7 +132,7 @@ def test_map_http_exception_from_structured_detail():
status_code, code, message, field = map_http_exception(exc)
assert status_code == status.HTTP_400_BAD_REQUEST
assert code == "validation_error"
assert code == status.HTTP_400_BAD_REQUEST
assert message == "bad input"
assert field == "username"
@ -141,7 +145,7 @@ def test_map_http_exception_unauthorized():
assert status_code == status.HTTP_401_UNAUTHORIZED
assert code == status.HTTP_401_UNAUTHORIZED
assert message == "Недействительный токен"
assert field is None
assert field == ""
def test_map_http_exception_not_found_maps_code_from_status():
@ -152,4 +156,4 @@ def test_map_http_exception_not_found_maps_code_from_status():
assert status_code == status.HTTP_404_NOT_FOUND
assert code == status.HTTP_404_NOT_FOUND
assert message == "not found"
assert field is None
assert field == ""

24
api/tests/unit/test_exception_handlers_http_shape.py
View File

@ -29,6 +29,10 @@ def _make_app() -> FastAPI:
exc.orig = Exception("budget_form #99 не существует")
raise exc
@app.get("/form/{form_id}")
async def form_by_id(form_id: int):
return {"form_id": form_id}
return app
@ -39,7 +43,7 @@ def test_http_exception_returns_structured_payload():
assert response.status_code == status.HTTP_401_UNAUTHORIZED
assert response.json() == {
"code": status.HTTP_401_UNAUTHORIZED,
"field": None,
"field": "",
"message": "Недействительный токен",
}
@ -51,7 +55,7 @@ def test_http_validation_exception_extracts_field():
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert response.json() == {
"code": status.HTTP_422_UNPROCESSABLE_ENTITY,
"field": None,
"field": "",
"message": "unknown_column: q5.plan",
}
@ -62,7 +66,19 @@ def test_sqlalchemy_exception_returns_structured_payload():
assert response.status_code == status.HTTP_404_NOT_FOUND
assert response.json() == {
"code": "not_found",
"field": None,
"code": status.HTTP_404_NOT_FOUND,
"field": "",
"message": "budget_form #99 не существует",
}
def test_request_validation_422_returns_structured_payload():
client = TestClient(_make_app())
response = client.get("/form/not-int")
assert response.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY
assert response.json() == {
"code": status.HTTP_422_UNPROCESSABLE_ENTITY,
"field": "form_id",
"message": "Input should be a valid integer, unable to parse string as an integer",
}

2
api/tests/unit/test_export_normalizers.py
View File

@ -1,4 +1,4 @@
from src.services.export_normalizers import (
from src.services.export_service.export_normalizers import (
ExcelValueSerializer,
Form1RowNormalizer,
Form3ReportRowNormalizer,

4
api/tests/unit/test_export_writers.py
View File

@ -3,11 +3,11 @@ from unittest.mock import AsyncMock
from openpyxl import Workbook
from src.services.export_service import ExportService
from src.services.export_writers import (
from src.services.export_service.export_mappings import (
FORM1_FLAT_COLUMNS,
SMETA_COLUMNS,
TabularSheetWriter,
)
from src.services.export_service.export_writers import TabularSheetWriter
def test_tabular_sheet_writer_writes_headers_and_rows():