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

Merge pull request 'add and fix validations on forms' (#10) from future/AURORA-1009 into test

Browse Source 
Reviewed-on: #10
Reviewed-by: tsygankoviva <tsygankov.itis@gmail.com>
This commit is contained in:
Raykov-MS 2026-05-18 14:27:43 +03:00
commit 7e9fc08056
5 changed files with 136 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
api/src/api/v1/forms.py
View File

@ -15,6 +15,8 @@ from src.api.v1.deps import get_current_active_user_with_set_db
from src.db.session import get_db
router = APIRouter(prefix="/form", tags=["forms"])
SHEETS_WITH_SECTIONS = {"AHR", "CAP", "OPER"}
FORM1_DIRECTION_REQUIRED_SHEETS = {"AHR", "CAP"}
def _rows_to_sheet_response(result: list[tuple]) -> list[SheetResponse]:
@ -28,6 +30,31 @@ def _rows_to_sheet_response(result: list[tuple]) -> list[SheetResponse]:
]
def _parse_sections_csv(sections: Optional[str]) -> Optional[list[str]]:
if sections is None:
return None
parsed = [section.strip() for section in sections.split(",") if section.strip()]
return parsed or None
def _validate_sheet_query_params(
form_type_code: FormTypeEnum,
sheet: str,
direction: Optional[DirectionSchemaEnum],
sections: Optional[list[str]],
) -> None:
if form_type_code == FormTypeEnum.FORM_1 and sheet in FORM1_DIRECTION_REQUIRED_SHEETS and not direction:
raise HTTPException(400, "direction обязателен для FORM_1 листов AHR/CAP")
if direction and not (
form_type_code == FormTypeEnum.FORM_1 and sheet in FORM1_DIRECTION_REQUIRED_SHEETS
):
raise HTTPException(400, "direction допустим только для FORM_1 листов AHR/CAP")
if sections and sheet not in SHEETS_WITH_SECTIONS:
raise HTTPException(400, f"sections не поддерживается для листа {sheet}")
@router.get("/")
async def get_forms(
offset: int = 0,
@ -51,7 +78,7 @@ async def get_forms(
)
@router.get("{form_id}/sheets")
@router.get("/{form_id}/sheets")
async def get_form_sheets(
form_id: int,
db: AsyncSession = Depends(get_db),
@ -81,8 +108,7 @@ async def get_sheet(
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> BaseListResponse[SheetResponse]:
if sections is not None:
sections = sections.split(",")
sections = _parse_sections_csv(sections)
t0 = time.perf_counter()
bf_service = BudgetFormService(db)
sheet_service = SheetService(db)
@ -96,10 +122,9 @@ async def get_sheet(
raise HTTPException(404, f"Форма {form_id} не найдена")
if sheet not in form.form_type.sheet_list:
raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
_validate_sheet_query_params(form.form_type.code, sheet, direction, sections)
if sections and (rem_sects := set(sections) - set(form.form_type.section_list)):
raise HTTPException(404, f"Секций {', '.join(rem_sects)} формы {form_id} не найден")
if form.form_type.code == FormTypeEnum.FORM_1 and not direction:
raise HTTPException(400, f"Форма типа {form.form_type.code} должна содержать в запросе direction")
raise HTTPException(400, f"Недопустимые sections: {', '.join(sorted(rem_sects))}")
result = await sheet_service.get(
form_id=form_id,
@ -126,8 +151,7 @@ async def update_cell(
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> BaseListResponse[SheetResponse]:
if sections is not None:
sections = sections.split(",")
sections = _parse_sections_csv(sections)
t0 = time.perf_counter()
bf_service = BudgetFormService(db)
sheet_service = SheetService(db)
@ -141,8 +165,9 @@ async def update_cell(
raise HTTPException(404, f"Форма {form_id} не найдена")
if sheet not in form.form_type.sheet_list:
raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
_validate_sheet_query_params(form.form_type.code, sheet, direction, sections)
if sections and (rem_sects := set(sections) - set(form.form_type.section_list)):
raise HTTPException(404, f"Секций {', '.join(rem_sects)} формы {form_id} не найден")
raise HTTPException(400, f"Недопустимые sections: {', '.join(sorted(rem_sects))}")
budget_line_service = BudgetLineService(db)
budget_line = await budget_line_service.get(budget_line_id=cell_body.line_id, user=current_user)
@ -181,8 +206,7 @@ async def update_cells(
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
) -> BaseListResponse[SheetResponse]:
if sections is not None:
sections = sections.split(",")
sections = _parse_sections_csv(sections)
t0 = time.perf_counter()
bf_service = BudgetFormService(db)
sheet_service = SheetService(db)
@ -196,8 +220,9 @@ async def update_cells(
raise HTTPException(404, f"Форма {form_id} не найдена")
if sheet not in form.form_type.sheet_list:
raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
_validate_sheet_query_params(form.form_type.code, sheet, direction, sections)
if sections and (rem_sects := set(sections) - set(form.form_type.section_list)):
raise HTTPException(404, f"Секций {', '.join(rem_sects)} формы {form_id} не найден")
raise HTTPException(400, f"Недопустимые sections: {', '.join(sorted(rem_sects))}")
budget_line_service = BudgetLineService(db)
budget_lines = await budget_line_service.get_list(
@ -246,6 +271,7 @@ async def add_line(
raise HTTPException(404, f"Форма {form_id} не найдена")
if sheet not in form.form_type.sheet_list:
raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
_validate_sheet_query_params(form.form_type.code, sheet, body.direction, None)
result = await sheet_service.add_line(
form_id=form_id,
@ -278,7 +304,7 @@ async def delete_line(
sheet: str,
row_id: int,
response: Response,
direction: str | None = None,
direction: Optional[DirectionSchemaEnum] = None,
db: AsyncSession = Depends(get_db),
current_user: AppUser = Depends(get_current_active_user_with_set_db),
):
@ -295,12 +321,13 @@ async def delete_line(
raise HTTPException(404, f"Форма {form_id} не найдена")
if sheet not in form.form_type.sheet_list:
raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
_validate_sheet_query_params(form.form_type.code, sheet, direction, None)
result = await sheet_service.delete_line(
form_id=form_id,
sheet=sheet,
row_id=row_id,
direction=direction,
direction=direction.value if direction else None,
user=current_user,
)
db_ms = (time.perf_counter() - t0) * 1000

7
api/src/api/v1/projects.py
View File

@ -21,12 +21,17 @@ from src.services.project_service import ProjectService
router = APIRouter(tags=["forms"])
FORM3_ALLOWED_SECTIONS = {"q1", "q2", "q3", "q4", "year"}
def _parse_sections(sections: Optional[str]) -> Optional[list[str]]:
if not sections:
return None
return [s.strip() for s in sections.split(",") if s.strip()]
parsed = [s.strip() for s in sections.split(",") if s.strip()]
invalid = sorted(set(parsed) - FORM3_ALLOWED_SECTIONS)
if invalid:
raise HTTPException(400, f"Недопустимые sections для FORM_3: {', '.join(invalid)}")
return parsed
def _rows_to_payload(rows: list[tuple]) -> list[dict]:

16
api/src/repository/sheet_repository.py
View File

@ -32,14 +32,14 @@ class SheetRepository:
self,
form_id: int,
sheet: str,
direction: str,
direction: str | None,
sections: list[str] | None = None,
) -> list[tuple]:
if any(
[
not isinstance(form_id, int),
sheet not in ACCEPTABLE_SHEETS,
direction not in [s.value for s in DirectionEnum],
direction is not None and direction not in [s.value for s in DirectionEnum],
sections and (set(sections) - ACCEPTABLE_SECTIONS),
]
):
@ -72,8 +72,8 @@ class SheetRepository:
self,
form_id: int,
sheet: str,
direction: str,
sections: list[str],
direction: str | None,
sections: list[str] | None,
line_id: int,
column: str,
value: str | int | float | None = None,
@ -83,7 +83,7 @@ class SheetRepository:
[
not isinstance(form_id, int),
sheet not in ACCEPTABLE_SHEETS,
direction not in [s.value for s in DirectionEnum],
direction is not None and direction not in [s.value for s in DirectionEnum],
sections and (set(sections) - ACCEPTABLE_SECTIONS),
]
):
@ -118,8 +118,8 @@ class SheetRepository:
self,
form_id: int,
sheet: str,
direction: str,
sections: list[str],
direction: str | None,
sections: list[str] | None,
changes: list[dict],
user_id: int = None,
) -> list[tuple]:
@ -127,7 +127,7 @@ class SheetRepository:
[
not isinstance(form_id, int),
sheet not in ACCEPTABLE_SHEETS,
direction not in [s.value for s in DirectionEnum],
direction is not None and direction not in [s.value for s in DirectionEnum],
sections and (set(sections) - ACCEPTABLE_SECTIONS),
]
):

14
api/src/services/sheet_service.py
View File

@ -15,7 +15,7 @@ class SheetService:
async def get(
self,
sheet: str,
direction: str,
direction: str | None,
sections: list[str] | None = None,
form_id: int | None = None,
form: BudgetForm | None = None
@ -31,8 +31,8 @@ class SheetService:
self,
form_id: int,
sheet: str,
direction: str,
sections: list[str],
direction: str | None,
sections: list[str] | None,
line_id: int,
column: str,
value: str | int | float | None = None,
@ -53,8 +53,8 @@ class SheetService:
self,
form_id: int,
sheet: str,
direction: str,
sections: list[str],
direction: str | None,
sections: list[str] | None,
changes: list[dict],
user: AppUser | None = None,
) -> list[tuple]:
@ -71,8 +71,8 @@ class SheetService:
self,
form_id: int,
sheet: str,
direction: str,
sections: list[str],
direction: str | None,
sections: list[str] | None,
line_id: int,
column: str,
value: str | int | float | None = None,

74
api/tests/unit/test_query_validation.py Normal file
View File

@ -0,0 +1,74 @@
import pytest
from fastapi import HTTPException
from src.api.v1.forms import _parse_sections_csv, _validate_sheet_query_params
from src.api.v1.projects import _parse_sections
from src.db.models.form_type import FormTypeEnum
from src.domain.schemas import DirectionSchemaEnum
def test_parse_sections_csv_returns_none_for_empty_values():
assert _parse_sections_csv(None) is None
assert _parse_sections_csv("") is None
assert _parse_sections_csv(" , , ") is None
def test_parse_sections_csv_splits_and_trims():
assert _parse_sections_csv("plan, q1, q2 ") == ["plan", "q1", "q2"]
def test_validate_sheet_query_requires_direction_for_form1_ahr_cap():
with pytest.raises(HTTPException) as exc:
_validate_sheet_query_params(FormTypeEnum.FORM_1, "AHR", None, None)
assert exc.value.status_code == 400
def test_validate_sheet_query_rejects_direction_for_non_form1_or_non_ahr_cap():
with pytest.raises(HTTPException) as exc1:
_validate_sheet_query_params(
FormTypeEnum.FORM_2,
"AHR",
DirectionSchemaEnum.SUPPORT,
None,
)
assert exc1.value.status_code == 400
with pytest.raises(HTTPException) as exc2:
_validate_sheet_query_params(
FormTypeEnum.FORM_1,
"SMETA",
DirectionSchemaEnum.SUPPORT,
None,
)
assert exc2.value.status_code == 400
def test_validate_sheet_query_rejects_sections_for_non_main_sheets():
with pytest.raises(HTTPException) as exc:
_validate_sheet_query_params(FormTypeEnum.FORM_1, "SMETA", None, ["q1"])
assert exc.value.status_code == 400
def test_validate_sheet_query_accepts_valid_combinations():
_validate_sheet_query_params(
FormTypeEnum.FORM_1,
"AHR",
DirectionSchemaEnum.SUPPORT,
["plan", "q1"],
)
_validate_sheet_query_params(
FormTypeEnum.FORM_2,
"AHR",
None,
["plan", "q1"],
)
def test_parse_form3_sections_accepts_supported_values():
assert _parse_sections("q1,q2,year") == ["q1", "q2", "year"]
def test_parse_form3_sections_rejects_invalid_values():
with pytest.raises(HTTPException) as exc:
_parse_sections("q1,q5")
assert exc.value.status_code == 400