320 lines
15 KiB
Python
320 lines
15 KiB
Python
"""AURORA-1848: тяжёлые GET отдают готовый JSON из Postgres.
|
||
|
||
Эталон — старый путь (методы репозиториев и сервисов, которыми пользуется выгрузка Excel)
|
||
в той же сессии теста. Спека: openspec/changes/aurora-1848-sheet-json/specs/sheet-read-api/spec.md
|
||
"""
|
||
import json
|
||
import uuid
|
||
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
|
||
from src.main import app
|
||
from src.repository.project_repository import ProjectRepository
|
||
from src.repository.sheet_repository import SheetRepository
|
||
from src.repository.summary_repository import SummaryRepository
|
||
from src.services.project_service import ProjectService
|
||
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
|
||
|
||
REPORT_NOT_FOUND = (400, {"code": 400, "field": "", "message": "Отчёт не найден"})
|
||
RF_ROLLUP_SQL_ERROR = (400, {"code": 400, "field": "", "message": "<class 'asyncpg.exceptions.UndefinedColumnError'>: column r.year does not exist"})
|
||
RF_FOREIGN = (403, {"code": 403, "field": "", "message": "Доступ запрещен"})
|
||
|
||
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))
|
||
|
||
|
||
def _create_project(client, headers, year=YEAR, branch_id=1) -> int:
|
||
r = client.post(
|
||
"/api/v1/projects",
|
||
json={
|
||
"name": f"PT_{uuid.uuid4().hex[:10]}",
|
||
"year": year,
|
||
"branch_id": branch_id,
|
||
"project_type": "Открытие ВСП",
|
||
"vsp_format": "Типовой",
|
||
"placement_type": "Собственность",
|
||
"object_address": "Тестовая улица 1",
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert r.status_code == 200, r.text
|
||
return r.json()["project_id"]
|
||
|
||
|
||
@pytest.fixture
|
||
def project_id(client, admin_tokens, auth_headers) -> int:
|
||
"""Новый проект сразу получает отчёт LIMIT со всеми строками статей (POST строки на нём даёт 400/409
|
||
и ломает транзакцию теста), так что отдельная строка не нужна."""
|
||
return _create_project(client, auth_headers(admin_tokens))
|
||
|
||
|
||
@pytest.mark.parametrize("sections", [None, "q1,year"])
|
||
def test_project_report_matches_old_path(client, db_session, admin_tokens, auth_headers, project_id, sections):
|
||
pid = project_id
|
||
result = get_ok(client, f"/api/v1/projects/{pid}/report/{YEAR}/LIMIT", auth_headers(admin_tokens),
|
||
{"sections": sections} if sections else None)
|
||
old = old_rows(client, lambda: ProjectService(db_session).get_report_rows(
|
||
project_id=pid, year=YEAR, report_type="LIMIT",
|
||
sections=sections.split(",") if sections else None, user=ADMIN))
|
||
assert result == old
|
||
assert any(row["row_type"] == "INPUT" for row in result)
|
||
|
||
|
||
def test_rf_rollup_error_same_as_old_path(client, db_session, admin_tokens, auth_headers, project_id):
|
||
"""v3.v_form3_rf_rollup_jsonb в миграции 0002_functions.sql ссылается на несуществующий r.year
|
||
(дефект вне AURORA-1848): на любой БД ответ - 400. Фиксируем, что новый путь отдаёт то же самое,
|
||
что и старый, а успешное сравнение с эталоном для rf-rollup невозможно."""
|
||
r = client.get(f"/api/v1/rf-rollup/1/{YEAR}", headers=auth_headers(admin_tokens))
|
||
assert (r.status_code, r.json()) == RF_ROLLUP_SQL_ERROR
|
||
|
||
|
||
def test_project_summary_matches_old_path(client, db_session, admin_tokens, auth_headers, project_id):
|
||
pid = project_id
|
||
result = get_ok(client, f"/api/v1/projects/{pid}/summary", auth_headers(admin_tokens))
|
||
old = old_rows(client, lambda: ProjectService(db_session).get_project_summary_rows(project_id=pid, user=ADMIN))
|
||
assert result == old
|
||
|
||
|
||
@pytest.mark.parametrize("kind", ["plan", "fact", "detail"])
|
||
def test_projects_summaries_match_old_path(client, db_session, admin_tokens, auth_headers, project_id, kind):
|
||
result = get_ok(client, f"/api/v1/projects/{kind}-summary", auth_headers(admin_tokens), {"year": YEAR})
|
||
method = getattr(ProjectService(db_session), f"get_{kind}_summary_rows")
|
||
old = old_rows(client, lambda: method(year=YEAR, branch_ids=None, user=ADMIN))
|
||
assert result == old
|
||
|
||
|
||
def test_projects_summaries_not_all_empty(client, admin_tokens, auth_headers, project_id):
|
||
pid = project_id
|
||
got = [get_ok(client, f"/api/v1/projects/{k}-summary", auth_headers(admin_tokens), {"year": YEAR})
|
||
for k in ("plan", "fact", "detail")]
|
||
got.append(get_ok(client, f"/api/v1/projects/{pid}/summary", auth_headers(admin_tokens)))
|
||
assert all(got), [bool(g) for g in got]
|
||
|
||
|
||
GET_METHODS = [
|
||
("/api/v1/projects/{pid}/report/%d/LIMIT" % YEAR, "get_report_rows"),
|
||
("/api/v1/rf-rollup/1/%d" % YEAR, "get_rf_rollup_rows"),
|
||
("/api/v1/projects/{pid}/summary", "get_project_summary_rows"),
|
||
("/api/v1/projects/plan-summary?year=%d" % YEAR, "get_plan_summary_rows"),
|
||
("/api/v1/projects/fact-summary?year=%d" % YEAR, "get_fact_summary_rows"),
|
||
("/api/v1/projects/detail-summary?year=%d" % YEAR, "get_detail_summary_rows"),
|
||
]
|
||
|
||
|
||
@pytest.mark.parametrize("url,method", GET_METHODS)
|
||
def test_projects_get_does_not_use_old_path(client, admin_tokens, auth_headers, project_id, monkeypatch, url, method):
|
||
called = []
|
||
|
||
async def boom(*args, **kwargs):
|
||
called.append(method)
|
||
raise AssertionError(f"GET не должен вызывать ProjectRepository.{method}")
|
||
monkeypatch.setattr(ProjectRepository, method, boom)
|
||
r = client.get(url.format(pid=project_id), headers=auth_headers(admin_tokens))
|
||
assert not called
|
||
if method != "get_rf_rollup_rows": # rf-rollup: см. test_rf_rollup_error_same_as_old_path
|
||
assert r.status_code == 200, r.text
|
||
|
||
|
||
# Характеризующие тесты ошибок: литералы сняты со старого кода (до правки роутеров).
|
||
def test_project_report_not_found_unchanged(client, admin_tokens, auth_headers, project_id):
|
||
pid = project_id
|
||
r = client.get(f"/api/v1/projects/{pid}/report/2099/LIMIT", headers=auth_headers(admin_tokens))
|
||
assert (r.status_code, r.json()) == REPORT_NOT_FOUND
|
||
|
||
|
||
def test_rf_rollup_foreign_branch_unchanged(client, isp_tokens, auth_headers):
|
||
r = client.get(f"/api/v1/rf-rollup/1/{YEAR}", headers=auth_headers(isp_tokens))
|
||
assert (r.status_code, r.json()) == RF_FOREIGN
|