diff --git a/api/tests/integration/sheet_contract.py b/api/tests/integration/sheet_contract.py index c239c39..acd55d4 100644 --- a/api/tests/integration/sheet_contract.py +++ b/api/tests/integration/sheet_contract.py @@ -20,3 +20,108 @@ def assert_patch_rows_match_sheet(patch_rows: list[dict], sheet_rows: list[dict] 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" + + +# --- AURORA-1850: эталон (фильтр 0050 поверх полного v_form_view) и хелперы SQL --- + +Row4 = tuple # (row_type, depth, sort_order, data::text) + +_STRIP_ALL = [ + "plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking", + "contract_summary", "allocation", "collegial", "ckk", + "sequestration", "reserve", "contract_detail", + "q1", "q2", "q3", "q4", "totals", + "fact_q1", "fact_q2", "fact_q3", "fact_q4", +] + +_SAT_SHEETS = ("AHR_RENT", "AHR_UTILITY", "AHR_SECURITY") + + +def strip_keys(sections: list[str] | None) -> list[str]: + """Ключи верхнего уровня исключённых разделов (как v_strip в 0050).""" + if sections is None: + return [] + return [k for k in _STRIP_ALL if k not in sections] + + +def sheet_kind(sheet: str) -> str: + if sheet in _SAT_SHEETS: + return "sat" + if sheet == "AHR_LIMIT": + return "limit" + if sheet == "OTCH9F": + return "otch9f" + return "tree" + + +_REF_FROM = """ + SELECT v.row_type, v.depth, v.sort_order, (v.data - CAST(:strip AS text[]))::text + FROM v3.v_form_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar)) v + WHERE {where} + ORDER BY v.sort_order""" + +_REF_WHERE = { + "sat": "v.row_type = 'INPUT' AND (v.data->>'id')::int = ANY(CAST(:ids AS int[]))", + "limit": "(v.data->>'id')::int = ANY(v3._limit_template_path_set(CAST(:ids AS int[])))", + "otch9f": """EXISTS (SELECT 1 FROM v3.fixed_asset_report far + WHERE far.id = ANY(CAST(:ids AS int[])) + AND far.expense_item_id = (v.data->>'expense_item_id')::int + AND far.equipment_name IS NOT DISTINCT FROM (v.data->>'equipment_name'))""", + "tree": """(v.row_type IN ('ROOT','GROUP','ITEM','SUB_ITEM','SUB_SUB_ITEM') + AND (v.data->'header'->>'expense_item_id')::int = ANY(v3._expense_item_path_set( + ARRAY(SELECT DISTINCT bl.expense_item_id FROM v3.budget_line bl + WHERE bl.id = ANY(CAST(:ids AS int[])))))) + OR (v.row_type = 'INPUT' AND (v.data->>'line_id')::int = ANY(CAST(:ids AS int[])))""", +} + + +def _rows(db_exec, sql: str, **params) -> list: + return [tuple(r) for r in db_exec(sql, **params)] + + +def reference_affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None, strip=True): + sql = _REF_FROM.format(where=_REF_WHERE[sheet_kind(sheet)]) + return _rows( + db_exec, sql, f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections, + strip=strip_keys(sections) if strip else [], + ) + + +def affected_rows(db_exec, form_id, sheet, ids, direction=None, sections=None): + return _rows( + db_exec, + """SELECT r.row_type, r.depth, r.sort_order, r.data::text + FROM v3._form_affected_rows(:f, :s, CAST(:ids AS int[]), CAST(:d AS varchar), + CAST(:sec AS text[])) r + ORDER BY r.sort_order""", + f=form_id, s=sheet, ids=list(ids), d=direction, sec=sections, + ) + + +def focus_builder_rows(db_exec, form_type, form_id, sheet, ids, direction=None, sections=None): + if form_type == "FORM_2": + sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text + FROM v3.v_form2_view(:f, :s, CAST(:sec AS text[]), CAST(:ids AS int[])) r + ORDER BY r.sort_order""" + params = dict(f=form_id, s=sheet, sec=sections, ids=list(ids)) + else: + sql = """SELECT r.row_type, r.depth, r.sort_order, r.data::text + FROM v3.v_form1_view(:f, :s, CAST(:sec AS text[]), CAST(:d AS varchar), + CAST(:ids AS int[])) r + ORDER BY r.sort_order""" + params = dict(f=form_id, s=sheet, sec=sections, d=direction, ids=list(ids)) + return _rows(db_exec, sql, **params) + + +def assert_rows_equal(actual: list, expected: list, label: str) -> None: + """Построчное сравнение; сообщение называет первую отличающуюся строку и длины.""" + if actual == expected: + return + n = min(len(actual), len(expected)) + first = next((i for i in range(n) if actual[i] != expected[i]), n) + a = actual[first] if first < len(actual) else None + e = expected[first] if first < len(expected) else None + raise AssertionError( + f"{label}: длины actual={len(actual)} expected={len(expected)}; " + f"первое расхождение в позиции {first}:\n actual = {a}\n expected = {e}" + ) diff --git a/api/tests/integration/test_upd_form_cells_subtree.py b/api/tests/integration/test_upd_form_cells_subtree.py new file mode 100644 index 0000000..19621ee --- /dev/null +++ b/api/tests/integration/test_upd_form_cells_subtree.py @@ -0,0 +1,412 @@ +"""AURORA-1850: эквивалентность пересчёта поддерева и полной сборки листа. + +Эталон — фильтр 0050 поверх полного v3.v_form_view. Новый путь (v3._form_affected_rows и +сборщики v3.v_form2_view / v3.v_form1_view с фокусом по id) обязан давать те же строки. +""" + +import uuid + +import pytest + +from sheet_contract import ( + affected_rows, + assert_patch_rows_match_sheet, + assert_rows_equal, + focus_builder_rows, + reference_affected_rows, + row_key, +) +from test_upd_form_cells_full_rows import ( # noqa: F401 (фикстуры и хелперы) + API, + _add_line, + _get_sheet, + _new_form, + _patch_cell, + _qs, + db_exec, + headers, +) + +TREE_TYPES = ("ROOT", "GROUP", "ITEM", "SUB_ITEM", "SUB_SUB_ITEM") +SATELLITES = ("AHR_RENT", "AHR_UTILITY", "AHR_SECURITY") +FORM2_SHEETS = ("AHR", "CAP", "OPER", "AHR_RENT", "AHR_UTILITY", "AHR_SECURITY", "AHR_LIMIT", "OTCH9F") + +# (form_type, sheet, direction, sections, год новой формы) +CASES = { + "F1-AHR-Support": ("FORM_1", "AHR", "Support", None, 2041), + "F1-CAP-Support": ("FORM_1", "CAP", "Support", None, 2043), + "F1-OPER": ("FORM_1", "OPER", None, None, 2044), + "F2-AHR": ("FORM_2", "AHR", None, None, 2045), + "F2-CAP": ("FORM_2", "CAP", None, None, 2046), + "F2-OPER": ("FORM_2", "OPER", None, None, 2047), + "F2-AHR_RENT": ("FORM_2", "AHR_RENT", None, None, 0), + "F2-AHR_UTILITY": ("FORM_2", "AHR_UTILITY", None, None, 0), + "F2-AHR_SECURITY": ("FORM_2", "AHR_SECURITY", None, None, 0), + "F2-AHR_LIMIT": ("FORM_2", "AHR_LIMIT", None, None, 0), + "F2-OTCH9F": ("FORM_2", "OTCH9F", None, None, 0), + "F2-AHR-plan": ("FORM_2", "AHR", None, ["plan"], 2048), +} +SAT_FORM = 4 + + +def _ordered_leaves(db_exec, form_type, sheet): + """Листовые статьи, привязанные к типу формы, в порядке mv_expense_item_tree.path.""" + return [ + (r[0], r[1], list(r[2])) + for r in db_exec( + """ + SELECT ei.id, ei.parent_id, t.path + FROM v3.expense_item ei + JOIN v3.mv_expense_item_tree t ON t.id = ei.id + JOIN v3.expense_item_form_type ft ON ft.expense_item_id = ei.id AND ft.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) + ORDER BY t.path + """, + ft=form_type, sheet=sheet, + ) + ] + + +def _pick_items(db_exec, form_type, sheet): + """первая, последняя, две сестры, строка из другого ROOT (без повторов).""" + leaves = _ordered_leaves(db_exec, form_type, sheet) + first, last = leaves[0], leaves[-1] + chosen = [first, last] + by_parent = {} + for lf in leaves: + by_parent.setdefault(lf[1], []).append(lf) + sib = next( + (g for g in by_parent.values() if len(g) >= 2 and all(x not in chosen for x in g[:2])), None + ) + sibs = sib[:2] if sib else [] + other = next( + (lf for lf in leaves if lf[2][0] not in (first[2][0], last[2][0]) and lf not in chosen + sibs), + None, + ) + items = [first[0], last[0]] + [s[0] for s in sibs] + ([other[0]] if other else []) + return items, {"first": first[0], "last": last[0], "sibs": [s[0] for s in sibs], + "other": other[0] if other else None} + + +def _input_line_ids(rows): + return [r["data"]["line_id"] for r in rows if r["row_type"] == "INPUT"] + + +def _prepare(client, headers, db_exec, case): + """Создаёт данные листа. Возвращает (form_id, units, batches).""" + form_type, sheet, direction, sections, year = CASES[case] + kind = "tree" if sheet in ("AHR", "CAP", "OPER") else sheet + if kind == "tree": + form_id = _new_form(client, headers, form_type, year) + items, names = _pick_items(db_exec, form_type, sheet) + lines = [] + for n, item in enumerate(items): + lines.append(_add_line(client, headers, form_id, sheet, item, direction)) + for n, line in enumerate(lines): + _patch_cell(client, headers, form_id, sheet, line, "plan.q1", 10 + n, direction) + _patch_cell(client, headers, form_id, sheet, lines[0], "header.justification", "обоснование", direction) + batch = list(lines) + if case == "F1-AHR-Support": + dev = _add_line(client, headers, form_id, sheet, items[0], "Development") + _patch_cell(client, headers, form_id, sheet, dev, "plan.q1", 77, "Development") + batch.append(dev) + get_rows = _get_sheet(client, headers, form_id, sheet, direction) + units = _input_line_ids(get_rows) + assert set(lines) <= set(units) + return form_id, units, [batch, batch + [lines[0]]] + if sheet in SATELLITES: + ids = [] + for n in range(3): + r = client.post( + f"{API}/{SAT_FORM}/sheet/{sheet}/line", + json={"vsp_id": 1 + n % 3, "contract_number": f"C-{uuid.uuid4().hex[:6]}"}, + headers=headers, + ) + assert r.status_code == 200, r.text + sid = r.json()["result"][0]["data"]["id"] + _patch_cell(client, headers, SAT_FORM, sheet, sid, "plan.q1", 5 + n) + ids.append(sid) + get_rows = _get_sheet(client, headers, SAT_FORM, sheet) + units = [r["data"]["id"] for r in get_rows if r["row_type"] == "INPUT"] + assert set(ids) <= set(units) + return SAT_FORM, units, [ids, ids + [ids[0]]] + if sheet == "AHR_LIMIT": + extra = [r[0] for r in db_exec( + "SELECT id FROM v3.limit_template WHERE row_type = 'LEAF' AND id <> 13 ORDER BY id LIMIT 2")] + ids = [13] + extra + for n, i in enumerate(ids): + _patch_cell(client, headers, SAT_FORM, sheet, i, "qty_q1", 3 + n) + units = [r[0] for r in db_exec("SELECT id FROM v3.limit_template WHERE row_type = 'LEAF' ORDER BY id")] + return SAT_FORM, units, [ids, ids + [ids[0]]] + if sheet == "OTCH9F": + return SAT_FORM, *_prepare_otch9f(client, headers, db_exec) + raise AssertionError(case) + + +def _prepare_otch9f(client, headers, db_exec): + items = [r[0] for r in db_exec( + """SELECT ei.id FROM v3.expense_item ei + WHERE ei.sheet = 'AHR' + AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = ei.id) + AND EXISTS (SELECT 1 FROM v3.expense_item_form_type t + WHERE t.expense_item_id = ei.id AND t.form_type_code = 'FORM_2') + AND NOT EXISTS (SELECT 1 FROM v3.fixed_asset_report far + WHERE far.budget_form_id = :f AND far.expense_item_id = ei.id) + ORDER BY ei.id LIMIT 2""", f=SAT_FORM)] + layout = [(items[0], "E1", 1), (items[0], "E1", 2), (items[0], "E1", 3), (items[0], "E2", 1), + (items[0], None, 1), (items[0], None, 2), (items[1], "E1", 1)] + ids = [] + for item, name, month in layout: + ids.append(db_exec( + """INSERT INTO v3.fixed_asset_report + (budget_form_id, expense_item_id, equipment_name, month, + opening_qty_604, opening_amt_604, acquired_qty_604, acquired_amt_604, + disposed_qty_604, disposed_amt_604, + opening_qty_60415, opening_amt_60415, acquired_qty_60415, acquired_amt_60415) + VALUES (:f, :i, :n, :m, 10, 1000, 2, 200, 1, 100, 5, 500, 1, 50) RETURNING id""", + f=SAT_FORM, i=item, n=name, m=month)[0][0]) + return ids, [ids, ids + [ids[0]]] + + +def _check(db_exec, path, form_type, form_id, sheet, ids, direction, sections, label): + if path == "affected": + actual = affected_rows(db_exec, form_id, sheet, ids, direction, sections) + expected = reference_affected_rows(db_exec, form_id, sheet, ids, direction, sections, strip=True) + else: + actual = focus_builder_rows(db_exec, form_type, form_id, sheet, ids, direction, sections) + expected = reference_affected_rows(db_exec, form_id, sheet, ids, direction, sections, strip=False) + assert_rows_equal(actual, expected, label) + + +@pytest.mark.parametrize("path", ["affected", "builder"]) +@pytest.mark.parametrize("case", [c for c in CASES]) +def test_affected_rows_match_reference(client, headers, db_exec, path, case): + form_type, sheet, direction, sections, _ = CASES[case] + form_id, units, batches = _prepare(client, headers, db_exec, case) + assert units, "в листе нет строк" + for i in units: + _check(db_exec, path, form_type, form_id, sheet, [i], direction, sections, f"{case} id={i}") + for n, batch in enumerate(batches): + _check(db_exec, path, form_type, form_id, sheet, batch, direction, sections, f"{case} batch#{n}") + + +EDIT_CASES = ["F1-AHR-Support", "F2-AHR", "F2-AHR_RENT", "F2-AHR_LIMIT", "F2-OTCH9F", "delta"] +VF_SIG = "v3.v_form_view(integer,character varying,text[],character varying,integer)" +_CALLS = "pg_stat_get_xact_function_calls(CAST(CAST('" + VF_SIG + "' AS regprocedure) AS oid))" + + +@pytest.mark.parametrize("case", EDIT_CASES) +def test_edit_path_does_not_call_v_form_view(client, headers, db_exec, case): + # R-2: роль perf — суперпользователь, track_functions доступен; тест не пропускается. + assert db_exec("SELECT rolsuper FROM pg_roles WHERE rolname = current_user")[0][0] is True, ( + "для track_functions нужна суперпользовательская роль" + ) + if case == "delta": + r = client.post( + f"{API}/{SAT_FORM}/sheet/AHR_RENT/line", + json={"vsp_id": 1, "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_cell(client, headers, SAT_FORM, "AHR_RENT", rent_id, "plan.q1", 100) + call, params = ("SELECT count(*) FROM v3.get_form2_ahr_satellite_delta(4, 'AHR_RENT', :i)"), {"i": rent_id} + else: + form_type, sheet, direction, sections, _ = CASES[case] + form_id, units, batches = _prepare(client, headers, db_exec, case) + call = ("SELECT count(*) FROM v3._form_affected_rows(:f, :s, CAST(:ids AS int[]), " + "CAST(:d AS varchar), CAST(:sec AS text[]))") + params = dict(f=form_id, s=sheet, ids=batches[0], d=direction, sec=sections) + db_exec("SET LOCAL track_functions = 'all'") + n0 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0] + db_exec(call, **params) + n1 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0] + assert n1 - n0 == 0, f"v_form_view вызван {n1 - n0} раз на пути правки" + + +def test_empty_focus_returns_nothing(db_exec): + for s in FORM2_SHEETS: + rows = db_exec( + "SELECT * FROM v3.v_form2_view(4, :s, CAST(NULL AS text[]), CAST('{}' AS int[]))", s=s) + assert rows == [], f"v_form2_view пуст для {s}: {len(rows)}" + rows = db_exec( + "SELECT * FROM v3.v_form1_view(1, 'AHR', CAST(NULL AS text[]), 'Support', CAST('{}' AS int[]))") + assert rows == [] + assert db_exec("SELECT * FROM v3._form_affected_rows(4, 'AHR', CAST('{}' AS int[]))") == [] + assert db_exec("SELECT * FROM v3.get_form2_ahr_satellite_delta(4, 'AHR_RENT', -1)") == [] + + +def _patch_cells(client, headers, form_id, sheet, changes, direction=None): + r = client.patch( + f"{API}/{form_id}/sheet/{sheet}/cells{_qs(direction)}", json={"changes": changes}, headers=headers) + assert r.status_code == 200, r.text + return r.json()["result"] + + +def test_patch_cells_two_branches_matches_get(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_2", 2054) + items, names = _pick_items(db_exec, "FORM_2", "AHR") + l1 = _add_line(client, headers, form_id, "AHR", names["first"]) + l2 = _add_line(client, headers, form_id, "AHR", names["last"]) + patch = _patch_cells(client, headers, form_id, "AHR", [ + {"line_id": l1, "column": "plan.q1", "value": 11}, + {"line_id": l2, "column": "plan.q1", "value": 22}, + ]) + get_rows = _get_sheet(client, headers, form_id, "AHR") + assert_patch_rows_match_sheet(patch, get_rows) + assert {r["data"]["line_id"] for r in patch if r["row_type"] == "INPUT"} == {l1, l2} + keys = [row_key(r) for r in patch] + assert len(keys) == len(set(keys)) + + +@pytest.mark.parametrize( + "form_type,direction,year", + [("FORM_2", None, 2055), ("FORM_1", "Support", 2056)], + ids=["FORM_2-AHR", "FORM_1-AHR-Support"], +) +def test_patch_after_add_and_delete_matches_get(client, headers, db_exec, form_type, direction, year): + form_id = _new_form(client, headers, form_type, year) + leaves = _ordered_leaves(db_exec, form_type, "AHR") + step = max(len(leaves) // 5, 1) + items = [leaves[i * step][0] for i in range(5)] + lines = [_add_line(client, headers, form_id, "AHR", it, direction) for it in items] + get0 = _get_sheet(client, headers, form_id, "AHR", direction) + order = [r["data"]["line_id"] for r in get0 if r["row_type"] == "INPUT" and r["data"]["line_id"] in lines] + victim, below = order[2], order[3] + r = client.delete(f"{API}/{form_id}/sheet/AHR/line/{victim}{_qs(direction)}", headers=headers) + assert r.status_code == 200, r.text + patch = _patch_cell(client, headers, form_id, "AHR", below, "plan.q1", 42, direction) + get_rows = _get_sheet(client, headers, form_id, "AHR", direction) + assert_patch_rows_match_sheet(patch, get_rows) + + def ident(row): + if row["row_type"] == "INPUT": + return ("L", row["data"]["line_id"]) + return ("N", row["data"]["header"]["expense_item_id"], row["row_type"]) + + by_ident = {ident(r): r["sort_order"] for r in get_rows if r["row_type"] != "META"} + for row in patch: + assert row["sort_order"] == by_ident[ident(row)], ident(row) + + +def test_patch_form1_development_matches_get(client, headers, db_exec): + form_id = _new_form(client, headers, "FORM_1", 2057) + items, names = _pick_items(db_exec, "FORM_1", "AHR") + sup = _add_line(client, headers, form_id, "AHR", names["first"], "Support") + dev1 = _add_line(client, headers, form_id, "AHR", names["first"], "Development") + dev2 = _add_line(client, headers, form_id, "AHR", names["last"], "Development") + _patch_cell(client, headers, form_id, "AHR", sup, "plan.q1", 5, "Support") + _patch_cell(client, headers, form_id, "AHR", dev2, "plan.q1", 6, "Development") + patch = _patch_cell(client, headers, form_id, "AHR", dev1, "plan.q1", 7, "Development") + assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR", "Development")) + assert {r["data"]["line_id"] for r in patch if r["row_type"] == "INPUT"} == {dev1} + + +def test_patch_otch9f_returns_all_months_of_equipment(client, headers, db_exec): + ids, _ = _prepare_otch9f(client, headers, db_exec) + e1_m1 = ids[0] + item = db_exec("SELECT expense_item_id FROM v3.fixed_asset_report WHERE id = :i", i=e1_m1)[0][0] + before = _get_sheet(client, headers, SAT_FORM, "OTCH9F") + patch = _patch_cell(client, headers, SAT_FORM, "OTCH9F", e1_m1, "b_604.acquired_qty", 9) + get_rows = _get_sheet(client, headers, SAT_FORM, "OTCH9F") + assert_patch_rows_match_sheet(patch, get_rows) + + def of_equipment(rows): + return {row_key(r): r for r in rows + if r["row_type"] != "META" and r["data"].get("expense_item_id") == item + and r["data"].get("equipment_name") == "E1"} + + expected = of_equipment(get_rows) + assert len(expected) >= 3 + assert {row_key(r) for r in patch} == set(expected) + prev = of_equipment(before) + changed = [k for k in expected if expected[k]["data"] != prev[k]["data"]] + assert len(changed) >= 2, "остатки следующих месяцев не пересчитаны" + + +def _sort_key_candidate(db_exec): + rows = db_exec( + """ + SELECT b.id, b.parent_id + FROM v3.mv_expense_item_tree b + JOIN v3.mv_expense_item_tree a ON a.id = b.parent_id + JOIN v3.expense_item_form_type fb ON fb.expense_item_id = b.id AND fb.form_type_code = 'FORM_2' + JOIN v3.expense_item_form_type fa ON fa.expense_item_id = a.id AND fa.form_type_code = 'FORM_2' + WHERE b.sheet = 'AHR' AND a.depth = 3 AND b.depth = 4 + AND NOT EXISTS (SELECT 1 FROM v3.expense_item c WHERE c.parent_id = b.id) + AND NOT EXISTS (SELECT 1 FROM v3.budget_line bl WHERE bl.id = b.id) + ORDER BY b.id LIMIT 1 + """ + ) + return rows[0][0] if rows else None + + +def test_sort_key_tie_node_before_line(client, headers, db_exec): + child = _sort_key_candidate(db_exec) + if child is None: + pytest.skip("нет статьи AHR глубины 3 с дочерней глубины 4 и свободным id в budget_line") + form_id = _new_form(client, headers, "FORM_2", 2058) + db_exec( + "INSERT INTO v3.budget_line (id, budget_form_id, expense_item_id, name) " + "VALUES (:i, :f, :i, 'tie')", i=child, f=form_id) + g1 = _get_sheet(client, headers, form_id, "AHR") + g2 = _get_sheet(client, headers, form_id, "AHR") + assert [row_key(r) for r in g1] == [row_key(r) for r in g2] + idx = next(n for n, r in enumerate(g1) if r["row_type"] == "INPUT" and r["data"]["line_id"] == child) + prev = g1[idx - 1] + assert prev["row_type"] in TREE_TYPES and prev["data"]["header"]["expense_item_id"] == child, prev + patch = _patch_cell(client, headers, form_id, "AHR", child, "plan.q1", 8) + assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR")) + + +def _delta_equals_get(db_exec, sheet, row_id): + delta = [tuple(r) for r in db_exec( + "SELECT row_type, depth, sort_order, data::text " + "FROM v3.get_form2_ahr_satellite_delta(4, :s, :i)", s=sheet, i=row_id)] + full = {r[2]: tuple(r) for r in db_exec( + "SELECT row_type, depth, sort_order, data::text " + "FROM v3.v_form_view(4, 'AHR', NULL, 'Support')")} + assert delta, "дельта пуста" + for row in delta: + assert row[2] in full, f"sort_order {row[2]} нет в полном AHR" + assert_rows_equal([row], [full[row[2]]], f"delta {sheet} sort_order={row[2]}") + return delta + + +def test_ahr_delta_matches_get_after_rent_edit(client, headers, db_exec): + r = client.post( + f"{API}/{SAT_FORM}/sheet/AHR_RENT/line", + json={"vsp_id": 1, "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_cell(client, headers, SAT_FORM, "AHR_RENT", rent_id, "plan.q1", 100) + delta = _delta_equals_get(db_exec, "AHR_RENT", rent_id) + import json + assert any( + row[0] == "INPUT" and (json.loads(row[3]).get("plan") or {}).get("q1") == 100 for row in delta + ), "в дельте нет INPUT с plan.q1 = 100" + + +def test_ahr_delta_includes_sub_sub_item_for_depth4_limit(client, headers, db_exec): + rows = db_exec( + """ + SELECT lt.id, t.id + FROM v3.limit_template lt + JOIN v3.mv_expense_item_tree t ON t.num_group_id = lt.expense_item_code + JOIN v3.expense_item_form_type ft ON ft.expense_item_id = t.id AND ft.form_type_code = 'FORM_2' + WHERE lt.row_type = 'LEAF' AND t.sheet = 'AHR' AND t.depth = 4 + ORDER BY lt.id LIMIT 1 + """ + ) + if not rows: + pytest.skip("нет LEAF шаблона лимитов с кодом статьи AHR глубины 4") + lt_id, eid = rows[0] + _patch_cell(client, headers, SAT_FORM, "AHR_LIMIT", lt_id, "qty_q1", 4) + delta = _delta_equals_get(db_exec, "AHR_LIMIT", lt_id) + import json + assert any( + row[0] == "SUB_SUB_ITEM" and json.loads(row[3])["header"]["expense_item_id"] == eid for row in delta + ), "в дельте нет SUB_SUB_ITEM статьи глубины 4"