test(sheet): контрактный тест ответа PATCH /cell с полными строками (AURORA-1844)
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
6855e9163a
commit
f0cac30622
22
api/tests/integration/sheet_contract.py
Normal file
22
api/tests/integration/sheet_contract.py
Normal file
@ -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"
|
||||
261
api/tests/integration/test_upd_form_cells_full_rows.py
Normal file
261
api/tests/integration/test_upd_form_cells_full_rows.py
Normal file
@ -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"
|
||||
Loading…
x
Reference in New Issue
Block a user