From f0cac30622521d012798ff4e476111f1f4ea6dd4 Mon Sep 17 00:00:00 2001 From: Andrey Epifancev Date: Wed, 28 Oct 2026 18:18:07 +0400 Subject: [PATCH] =?UTF-8?q?test(sheet):=20=D0=BA=D0=BE=D0=BD=D1=82=D1=80?= =?UTF-8?q?=D0=B0=D0=BA=D1=82=D0=BD=D1=8B=D0=B9=20=D1=82=D0=B5=D1=81=D1=82?= =?UTF-8?q?=20=D0=BE=D1=82=D0=B2=D0=B5=D1=82=D0=B0=20PATCH=20/cell=20?= =?UTF-8?q?=D1=81=20=D0=BF=D0=BE=D0=BB=D0=BD=D1=8B=D0=BC=D0=B8=20=D1=81?= =?UTF-8?q?=D1=82=D1=80=D0=BE=D0=BA=D0=B0=D0=BC=D0=B8=20(AURORA-1844)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 5.5 --- api/tests/integration/sheet_contract.py | 22 ++ .../test_upd_form_cells_full_rows.py | 261 ++++++++++++++++++ 2 files changed, 283 insertions(+) create mode 100644 api/tests/integration/sheet_contract.py create mode 100644 api/tests/integration/test_upd_form_cells_full_rows.py diff --git a/api/tests/integration/sheet_contract.py b/api/tests/integration/sheet_contract.py new file mode 100644 index 0000000..c239c39 --- /dev/null +++ b/api/tests/integration/sheet_contract.py @@ -0,0 +1,22 @@ +"""Переиспользуемая проверка контракта ответа на правку ячеек. + +Каждая строка ответа PATCH /cell(s) должна совпасть (data и sort_order) со строкой +того же (row_type, depth, sort_order) из последующего GET /sheet. +Используется в AURORA-1844 и далее (AURORA-1850: пересчёт только затронутого поддерева). +""" + + +def row_key(row: dict) -> tuple: + return (row["row_type"], row["depth"], row["sort_order"]) + + +def assert_patch_rows_match_sheet(patch_rows: list[dict], sheet_rows: list[dict]) -> None: + """Строки ответа на правку целиком совпадают со строками листа.""" + sheet_by_key = {row_key(r): r for r in sheet_rows} + assert patch_rows, "ответ на правку пуст" + keys = [row_key(r) for r in patch_rows] + assert len(keys) == len(set(keys)), "в ответе дубли строк" + for row in patch_rows: + key = row_key(row) + assert key in sheet_by_key, f"строки {key} нет в GET /sheet" + assert row["data"] == sheet_by_key[key]["data"], f"data строки {key} расходится с GET" diff --git a/api/tests/integration/test_upd_form_cells_full_rows.py b/api/tests/integration/test_upd_form_cells_full_rows.py new file mode 100644 index 0000000..b8b9089 --- /dev/null +++ b/api/tests/integration/test_upd_form_cells_full_rows.py @@ -0,0 +1,261 @@ +"""AURORA-1844: v3.upd_form_cells возвращает затронутые строки целиком.""" + +import uuid + +import pytest +from sqlalchemy import text + +from src.db.session import get_db +from src.main import app + +from sheet_contract import assert_patch_rows_match_sheet + +API = "/api/v1/form" + + +@pytest.fixture +def headers(admin_tokens, auth_headers): + return auth_headers(admin_tokens) + + +@pytest.fixture +def db_exec(client): + """Выполняет SQL в той же (откатываемой) сессии, что и HTTP-запросы клиента.""" + state = {} + + def run(sql: str, **params): + async def _go(): + if "db" not in state: + state["gen"] = app.dependency_overrides[get_db]() + state["db"] = await state["gen"].__anext__() + res = await state["db"].execute(text(sql), params) + return res.all() if res.returns_rows else [] + + return client.portal.call(_go) + + return run + + +def _qs(direction=None, sections=None): + params = [] + if direction: + params.append(f"direction={direction}") + if sections: + params.append(f"sections={sections}") + return ("?" + "&".join(params)) if params else "" + + +def _get_sheet(client, headers, form_id, sheet, direction=None, sections=None): + r = client.get(f"{API}/{form_id}/sheet/{sheet}{_qs(direction, sections)}", headers=headers) + assert r.status_code == 200, r.text + return r.json()["result"] + + +def _patch_cell(client, headers, form_id, sheet, line_id, column, value, direction=None, sections=None): + r = client.patch( + f"{API}/{form_id}/sheet/{sheet}/cell{_qs(direction, sections)}", + json={"line_id": line_id, "column": column, "value": value}, + headers=headers, + ) + assert r.status_code == 200, r.text + return r.json()["result"] + + +def _new_form(client, headers, form_type, year, org=1): + r = client.post( + f"{API}/", + json={"year": year, "form_type_code": form_type, "org_unit_ids": [org]}, + headers=headers, + ) + assert r.status_code == 200, r.text + return r.json()["result"][0]["id"] + + +def _leaf_items(db_exec, form_type, sheet, n): + """n листовых статей одного родителя (чтобы у строк был общий предок-агрегат).""" + rows = db_exec( + """ + SELECT ei.id FROM v3.expense_item ei + JOIN v3.expense_item_form_type t ON t.expense_item_id = ei.id AND t.form_type_code = :ft + WHERE ei.sheet = :sheet AND ei.direction = 'Support' + AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = ei.id) + AND ei.parent_id IN ( + SELECT ei2.parent_id FROM v3.expense_item ei2 + JOIN v3.expense_item_form_type t2 ON t2.expense_item_id = ei2.id AND t2.form_type_code = :ft + WHERE ei2.sheet = :sheet AND ei2.direction = 'Support' + AND NOT EXISTS (SELECT 1 FROM v3.expense_item c2 WHERE c2.parent_id = ei2.id) + GROUP BY ei2.parent_id HAVING count(*) >= :n + ORDER BY ei2.parent_id LIMIT 1) + ORDER BY ei.id LIMIT :n + """, + ft=form_type, sheet=sheet, n=n, + ) + assert len(rows) == n + return [r[0] for r in rows] + + +def _add_line(client, headers, form_id, sheet, item_id, direction=None, **extra): + body = {"expense_item_id": item_id, **extra} + if direction: + body["direction"] = direction + r = client.post(f"{API}/{form_id}/sheet/{sheet}/line{_qs(direction)}", json=body, headers=headers) + assert r.status_code == 200, r.text + for row in r.json()["result"]: + if row["row_type"] == "INPUT": + return row["data"]["line_id"] + raise AssertionError("в ответе add_line нет INPUT") + + +def _plan_q1(row): + return row["data"]["plan"]["q1"] + + +def _aggregates(rows): + return [r for r in rows if r["row_type"] != "INPUT"] + + +def _two_lines_case(client, headers, db_exec, form_type, form_id, sheet, direction=None): + items = _leaf_items(db_exec, form_type, sheet, 2) + l1 = _add_line(client, headers, form_id, sheet, items[0], direction) + l2 = _add_line(client, headers, form_id, sheet, items[1], direction) + _patch_cell(client, headers, form_id, sheet, l1, "plan.q1", 10, direction) + patch = _patch_cell(client, headers, form_id, sheet, l2, "plan.q1", 20, direction) + return l1, l2, patch + + +@pytest.mark.parametrize( + "form_type,sheet,direction,year", + [ + ("FORM_1", "AHR", "Support", 2031), + ("FORM_2", "AHR", None, 2032), + ("FORM_2", "CAP", None, 2033), + ("FORM_2", "OPER", None, 2034), + ], +) +def test_patch_cell_returns_full_rows_matching_get( + client, headers, db_exec, form_type, sheet, direction, year +): + form_id = _new_form(client, headers, form_type, year) + l1, l2, patch = _two_lines_case(client, headers, db_exec, form_type, form_id, sheet, direction) + get_rows = _get_sheet(client, headers, form_id, sheet, direction) + + assert_patch_rows_match_sheet(patch, get_rows) + + inputs = [r for r in patch if r["row_type"] == "INPUT"] + assert [r["data"]["line_id"] for r in inputs] == [l2] + aggs = _aggregates(patch) + assert aggs, "нет предков-агрегатов" + # общий предок видит сумму обеих строк (форма новая, других строк нет) + assert any(_plan_q1(r) == 30 for r in aggs) + assert all(_plan_q1(r) in (30, 20, 10) for r in aggs if _plan_q1(r) is not None) + + +def test_unchanged_ancestor_is_returned_in_full(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_2", 2035) + l1, l2, _ = _two_lines_case(client, headers, db_exec, "FORM_2", form_id, "AHR") + patch = _patch_cell(client, headers, form_id, "AHR", l2, "header.justification", "обоснование") + get_rows = _get_sheet(client, headers, form_id, "AHR") + assert_patch_rows_match_sheet(patch, get_rows) + # суммы предков не менялись, но строки пришли целиком + assert any(_plan_q1(r) == 30 for r in _aggregates(patch)) + + +def test_clearing_value_decreases_ancestor_sums(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_2", 2036) + l1, l2, _ = _two_lines_case(client, headers, db_exec, "FORM_2", form_id, "AHR") + patch = _patch_cell(client, headers, form_id, "AHR", l2, "plan.q1", None) + assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR")) + inp = next(r for r in patch if r["row_type"] == "INPUT") + assert "q1" in inp["data"]["plan"] and inp["data"]["plan"]["q1"] in (None, 0) + assert all(_plan_q1(r) == 10 for r in _aggregates(patch) if _plan_q1(r) is not None) + + +def test_sections_filter_drops_other_sections_keys(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_2", 2037) + items = _leaf_items(db_exec, "FORM_2", "AHR", 1) + line = _add_line(client, headers, form_id, "AHR", items[0]) + patch = _patch_cell(client, headers, form_id, "AHR", line, "plan.q1", 5, sections="plan") + assert patch + allowed = {"plan", "header", "line_id", "sort_path"} + excluded = {"seq_dfip", "seq_ssp", "approved", "contract", "booking", "q1", "q2", + "q3", "q4", "totals", "contract_summary", "allocation", "collegial", "ckk"} + for row in patch: + assert not (set(row["data"]) & excluded), set(row["data"]) & excluded + assert "plan" in row["data"] + del allowed + + +def test_bad_column_in_batch_fails_and_applies_nothing(client, headers): + # строка фикстуры budget_line 1 (FORM_1 / AHR / Support) + before = _get_sheet(client, headers, 1, "AHR", "Support") + r = client.patch( + f"{API}/1/sheet/AHR/cells?direction=Support", + json={"changes": [ + {"line_id": 1, "column": "plan.q1", "value": 7}, + {"line_id": 1, "column": "bad.scope", "value": 1}, + ]}, + headers=headers, + ) + # unknown_column сейчас маппится в 400 (как в test_db_functions_integration) + assert r.status_code in (400, 422), r.text + after = _get_sheet(client, headers, 1, "AHR", "Support") + assert before == after + + +def test_form4_returns_only_own_program_and_project(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_4", 2039) + items = _leaf_items(db_exec, "FORM_4", "AHR", 2) + sec = db_exec("SELECT section_code FROM v3.expense_item WHERE id = :i", i=items[0])[0][0] + ids = {} + for pname in ("A", "B"): + prog = db_exec( + "INSERT INTO v3.form4_project (name, level, parent_id, form_id, section_code) " + "VALUES (:n, 'program', NULL, :f, :s) RETURNING id", + n=f"prog{pname}", f=form_id, s=sec)[0][0] + prj = db_exec( + "INSERT INTO v3.form4_project (name, level, parent_id, form_id) " + "VALUES (:n, 'project', :p, :f) RETURNING id", + n=f"prj{pname}", p=prog, f=form_id)[0][0] + ids[pname] = (prog, prj) + lines = {} + for pname, item in zip(("A", "B"), items): + lines[pname] = _add_line(client, headers, form_id, "AHR", item, project_id=ids[pname][1]) + patch = _patch_cell(client, headers, form_id, "AHR", lines["A"], "plan.q1", 15) + get_rows = _get_sheet(client, headers, form_id, "AHR") + assert_patch_rows_match_sheet(patch, get_rows) + + prog_a, prj_a = ids["A"] + types = [r["row_type"] for r in patch] + assert "ROOT" in types and "GROUP" in types and "ITEM" in types + for r in patch: + h = r["data"].get("header") or {} + if r["row_type"] == "GROUP": + assert h["program_id"] == prog_a + if r["row_type"] == "ITEM": + assert h["project_id"] == prj_a + if r["row_type"] == "INPUT": + assert r["data"]["line_id"] == lines["A"] + # в листе есть чужие программа/проект, в ответе их нет + assert {r["data"]["header"].get("program_id") for r in get_rows if r["row_type"] == "GROUP"} >= { + ids["A"][0], ids["B"][0]} + + +def test_ahr_rent_full_row_and_synced_into_ahr(client, headers): + form_id = 4 + vsp = 1 + r = client.post( + f"{API}/{form_id}/sheet/AHR_RENT/line", + json={"vsp_id": vsp, "contract_number": f"C-{uuid.uuid4().hex[:6]}"}, + headers=headers, + ) + assert r.status_code == 200, r.text + rent_id = r.json()["result"][0]["data"]["id"] + patch = _patch_cell(client, headers, form_id, "AHR_RENT", rent_id, "plan.q1", 100) + assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR_RENT")) + assert len(patch) == 1 + assert patch[0]["data"]["id"] == rent_id + assert patch[0]["data"]["plan"]["q1"] == 100 + ahr = _get_sheet(client, headers, form_id, "AHR") + assert any( + x["row_type"] != "META" and (x["data"].get("plan") or {}).get("q1") == 100 for x in ahr + ), "синхронизированная сумма не видна в AHR"