"""AURORA-1848: тяжёлые GET отдают готовый JSON из Postgres. Эталон — старый путь (методы репозиториев и сервисов, которыми пользуется выгрузка Excel) в той же сессии теста. Спека: openspec/changes/aurora-1848-sheet-json/specs/sheet-read-api/spec.md """ import json from types import SimpleNamespace import pytest from sqlalchemy import text from src.main import app from src.repository.sheet_repository import SheetRepository from src.repository.summary_repository import SummaryRepository from src.services.summary_service import SummaryService ADMIN = SimpleNamespace(id=1, role_id=1) # fixture.sql: admin ISP = SimpleNamespace(id=2, role_id=2) # fixture.sql: isp1, EXECUTOR_DFIP YEAR = 2026 HEAVY_GET_PATHS = [ "/api/v1/form/{form_id}/sheet/{sheet}", "/api/v1/svod/1/{sheet}/{year}", "/api/v1/svod/2/{sheet}/{year}", "/api/v1/projects/{project_id}/report/{year}/{report_type}", "/api/v1/rf-rollup/{branch_id}/{year}", "/api/v1/projects/{project_id}/summary", "/api/v1/projects/plan-summary", "/api/v1/projects/fact-summary", "/api/v1/projects/detail-summary", ] OPENAPI_REF = "#/components/schemas/BaseListResponse_SheetResponse_" def _payload(rows) -> list[dict]: """Строки старого пути в том виде, в каком их отдавал BaseListResponse[SheetResponse].""" return [{"row_type": r[0], "depth": r[1], "sort_order": r[2], "data": r[3]} for r in rows] def old_rows(client, coro_factory) -> list[dict]: """Вызвать старый метод (coro_factory() -> awaitable list[tuple]) в цикле приложения.""" async def _go(): return _payload(await coro_factory()) return client.portal.call(_go) def expected_sheet(rows: list[dict]) -> list[dict]: """Ожидаемый ответ листа и сводов по эталону старого пути. Фаза B (Task 13) меняет тело на strip_nulls.""" return rows def get_ok(client, url, headers, params=None) -> list[dict]: r = client.get(url, headers=headers, params=params) assert r.status_code == 200, r.text assert r.headers["content-type"] == "application/json" assert float(r.headers["x-db-time-ms"]) >= 0 body = r.json() assert list(body) == ["success", "message", "result", "count"] assert body["success"] is True and body["message"] is None assert body["count"] == len(body["result"]) for row in body["result"]: assert row.keys() == {"row_type", "depth", "sort_order", "data"} return body["result"] def test_openapi_heavy_get_schema_unchanged(): spec = app.openapi() for path in HEAVY_GET_PATHS: schema = spec["paths"][path]["get"]["responses"]["200"]["content"]["application/json"]["schema"] assert schema == {"$ref": OPENAPI_REF}, path SHEET_CASES = [ (1, "AHR", "Support", None), # FORM_1, лист с direction (4, "AHR", None, None), # FORM_2, без direction (4, "AHR", None, ["plan", "q1"]), # FORM_2, с sections ] def _sheet_params(direction, sections): params = {} if direction: params["direction"] = direction if sections: params["sections"] = ",".join(sections) return params @pytest.mark.parametrize("form_id,sheet,direction,sections", SHEET_CASES) def test_sheet_matches_old_path(client, db_session, admin_tokens, auth_headers, form_id, sheet, direction, sections): result = get_ok(client, f"/api/v1/form/{form_id}/sheet/{sheet}", auth_headers(admin_tokens), _sheet_params(direction, sections)) old = old_rows(client, lambda: SheetRepository(db_session).get(form_id=form_id, sheet=sheet, direction=direction, sections=sections)) assert result assert result[0]["row_type"] == "META" and result[0]["depth"] == -1 assert result == expected_sheet(old) TRICKY = 'Кавычка " и \\ слэш\nперенос\tтаб' def test_sheet_text_escaping_matches_old_path(client, db_session, admin_tokens, auth_headers): # В фикстуре у FORM_2 нет budget_line: вставляем строку с TRICKY в name (header.name листа). # justification в ответе листа не выводится; вставка откатывается вместе с сессией теста. async def _set(): await db_session.execute( text( "INSERT INTO v3.budget_line (budget_form_id, expense_item_id, name, direction) " "SELECT 4, t.id, :t, 'Support' FROM v3.mv_expense_item_tree t " "JOIN v3.expense_item_form_type eft ON eft.expense_item_id = t.id " "WHERE eft.form_type_code = 'FORM_2' AND t.sheet = 'AHR' ORDER BY t.id LIMIT 1" ), {"t": TRICKY}, ) client.portal.call(_set) result = get_ok(client, "/api/v1/form/4/sheet/AHR", auth_headers(admin_tokens)) old = old_rows(client, lambda: SheetRepository(db_session).get(form_id=4, sheet="AHR", direction=None)) assert result == expected_sheet(old) assert any((r["data"].get("header") or {}).get("name") == TRICKY for r in result) def test_sheet_get_does_not_use_old_path(client, admin_tokens, auth_headers, monkeypatch): async def boom(*args, **kwargs): raise AssertionError("GET листа не должен вызывать SheetRepository.get") monkeypatch.setattr(SheetRepository, "get", boom) get_ok(client, "/api/v1/form/4/sheet/AHR", auth_headers(admin_tokens)) SVOD2_PLAN_LINE_SQL = text( "WITH bf AS (INSERT INTO v3.budget_form (form_type_code, year, org_unit_id) " "VALUES ('FORM_1', :year, 2) RETURNING id), " "bl AS (INSERT INTO v3.budget_line (budget_form_id, expense_item_id) " "SELECT id, :expense_item_id FROM bf RETURNING id) " "INSERT INTO v3.plan (line_id, plan_q1) SELECT id, 1000 FROM bl" ) @pytest.fixture def svod2_plan_line(client, db_session): """Фикстура не даёт свода 2 строк за YEAR: вставляем FORM_1 с ненулевым планом. Откатывается вместе с сессией теста (POST строки без плана свод 2 не учитывает).""" async def _set(): await db_session.execute(SVOD2_PLAN_LINE_SQL, {"year": YEAR, "expense_item_id": 88}) client.portal.call(_set) @pytest.mark.parametrize("sheet", ["FORM_1", "FORM_2", "FORM_3", "FORM_4", "MAIN"]) def test_svod1_matches_old_path(client, db_session, admin_tokens, auth_headers, sheet): result = get_ok(client, f"/api/v1/svod/1/{sheet}/{YEAR}", auth_headers(admin_tokens)) old = old_rows(client, lambda: SummaryRepository(db_session).get_svod1_rows(year=YEAR, sheet=sheet, org_ids=None)) assert result == expected_sheet(old) def test_svod1_org_ids_matches_old_path(client, db_session, admin_tokens, auth_headers): result = get_ok(client, f"/api/v1/svod/1/FORM_1/{YEAR}", auth_headers(admin_tokens), {"org_ids": "1,2,3"}) old = old_rows(client, lambda: SummaryRepository(db_session).get_svod1_rows(year=YEAR, sheet="FORM_1", org_ids=[1, 2, 3])) assert result assert result == expected_sheet(old) SVOD2_CASES = [ ("GO", {}, {}), ("RF", {}, {}), ("GO", {"org_ids": "1,2"}, {"org_ids": [1, 2]}), ("GO", {"offset": 1, "limit": 2}, {"offset": 1, "limit": 2}), ("RF", {"sort_by": "section_code", "sort_direction": "desc"}, {"sort_by": "section_code", "sort_direction": "desc"}), ] @pytest.mark.usefixtures("svod2_plan_line") @pytest.mark.parametrize("sheet,query,repo_kwargs", SVOD2_CASES) def test_svod2_matches_old_path(client, db_session, admin_tokens, auth_headers, sheet, query, repo_kwargs): result = get_ok(client, f"/api/v1/svod/2/{sheet}/{YEAR}", auth_headers(admin_tokens), query) old = old_rows(client, lambda: SummaryRepository(db_session).get_svod2_rows(year=YEAR, sheet=sheet, **repo_kwargs)) assert result == expected_sheet(old) @pytest.mark.usefixtures("svod2_plan_line") def test_svod2_fixture_not_empty(client, admin_tokens, auth_headers): """Иначе сравнения свода 2 выше проходят на пустых списках.""" go = get_ok(client, f"/api/v1/svod/2/GO/{YEAR}", auth_headers(admin_tokens)) rf = get_ok(client, f"/api/v1/svod/2/RF/{YEAR}", auth_headers(admin_tokens)) assert go or rf @pytest.mark.usefixtures("svod2_plan_line") def test_svod2_isp_matches_service_old_path(client, db_session, isp_tokens, auth_headers): result = get_ok(client, f"/api/v1/svod/2/GO/{YEAR}", auth_headers(isp_tokens)) old = old_rows(client, lambda: SummaryService(db_session).get_svod2_rows(year=YEAR, sheet="GO", user=ISP)) assert result == expected_sheet(old) def test_svod2_foreign_org_ids_empty_body(client, isp_tokens, auth_headers): r = client.get(f"/api/v1/svod/2/GO/{YEAR}", headers=auth_headers(isp_tokens), params={"org_ids": "999999"}) assert r.status_code == 200 assert r.json() == {"success": True, "message": None, "result": [], "count": 0} assert float(r.headers["x-db-time-ms"]) >= 0 @pytest.mark.parametrize("url,method", [ (f"/api/v1/svod/1/MAIN/{YEAR}", "get_svod1_rows"), (f"/api/v1/svod/2/GO/{YEAR}", "get_svod2_rows"), ]) def test_svod_get_does_not_use_old_path(client, admin_tokens, auth_headers, monkeypatch, url, method): async def boom(*args, **kwargs): raise AssertionError(f"GET свода не должен вызывать SummaryRepository.{method}") monkeypatch.setattr(SummaryRepository, method, boom) get_ok(client, url, auth_headers(admin_tokens))