diff --git a/api/tests/integration/sheet_contract.py b/api/tests/integration/sheet_contract.py index acd55d4..f93ea17 100644 --- a/api/tests/integration/sheet_contract.py +++ b/api/tests/integration/sheet_contract.py @@ -24,7 +24,7 @@ def assert_patch_rows_match_sheet(patch_rows: list[dict], sheet_rows: list[dict] # --- AURORA-1850: эталон (фильтр 0050 поверх полного v_form_view) и хелперы SQL --- -Row4 = tuple # (row_type, depth, sort_order, data::text) +Row4 = tuple[str, int, int, str] # (row_type, depth, sort_order, data::text) _STRIP_ALL = [ "plan", "seq_dfip", "seq_ssp", "approved", "contract", "booking", diff --git a/api/tests/integration/test_upd_form_cells_subtree.py b/api/tests/integration/test_upd_form_cells_subtree.py index 19621ee..faa850c 100644 --- a/api/tests/integration/test_upd_form_cells_subtree.py +++ b/api/tests/integration/test_upd_form_cells_subtree.py @@ -80,10 +80,15 @@ def _pick_items(db_exec, form_type, sheet): (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, - ) + assert len(sibs) == 2, "нет пары сестёр, отличных от первой/последней" + n_roots = len({lf[2][0] for lf in leaves}) + other = None + if n_roots >= 2: + assert first[2][0] != last[2][0], "первая и последняя статьи в одном ROOT при нескольких ROOT" + # строка из ROOT, отличного от ROOT сестёр + other = next((lf for lf in leaves + if lf[2][0] != sibs[0][2][0] and lf not in chosen + sibs), None) + assert other is not None, "нет строки из другого ROOT" 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} @@ -169,14 +174,25 @@ def _prepare_otch9f(client, headers, db_exec): return ids, [ids, ids + [ids[0]]] -def _check(db_exec, path, form_type, form_id, sheet, ids, direction, sections, label): +def _check(db_exec, path, form_type, form_id, sheet, ids, direction, sections, label, must_have_input=False): 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 expected, f"{label}: empty reference" assert_rows_equal(actual, expected, label) + if must_have_input: + assert any(_has_input(r, ids[0]) for r in actual), f"{label}: нет INPUT id={ids[0]} в результате" + + +def _has_input(row, i): + import json + if row[0] != "INPUT": + return False + d = json.loads(row[3]) + return d.get("line_id") == i or d.get("id") == i @pytest.mark.parametrize("path", ["affected", "builder"]) @@ -185,8 +201,15 @@ 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, "в листе нет строк" + is_tree = sheet in ("AHR", "CAP", "OPER") for i in units: - _check(db_exec, path, form_type, form_id, sheet, [i], direction, sections, f"{case} id={i}") + _check(db_exec, path, form_type, form_id, sheet, [i], direction, sections, f"{case} id={i}", + must_have_input=sheet != "AHR_LIMIT" and sheet != "OTCH9F") + if case == "F1-AHR-Support" and path == "affected": + dev = batches[0][-1] + rows = affected_rows(db_exec, form_id, sheet, [dev], direction, sections) + assert not any(_has_input(r, dev) for r in rows), "Development INPUT попал в результат Support" + del is_tree for n, batch in enumerate(batches): _check(db_exec, path, form_type, form_id, sheet, batch, direction, sections, f"{case} batch#{n}") @@ -219,7 +242,11 @@ def test_edit_path_does_not_call_v_form_view(client, headers, db_exec, case): "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] + c0 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0] + db_exec("SELECT count(*) FROM v3.v_form_view(4, 'AHR_RENT', NULL, NULL)") + c1 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0] + assert c1 - c0 == 1, "положительный контроль: счётчик не отслеживает v_form_view" + n0 = c1 db_exec(call, **params) n1 = db_exec(f"SELECT COALESCE({_CALLS}, 0)")[0][0] assert n1 - n0 == 0, f"v_form_view вызван {n1 - n0} раз на пути правки" @@ -258,6 +285,13 @@ def test_patch_cells_two_branches_matches_get(client, headers, db_exec): 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)) + roots = { + db_exec("SELECT path[1] FROM v3.mv_expense_item_tree WHERE id = :i", i=i)[0][0] + for i in (names["first"], names["last"]) + } + assert len(roots) == 2, "строки l1/l2 в одном ROOT" + root_eids = {r["data"]["header"]["expense_item_id"] for r in patch if r["row_type"] == "ROOT"} + assert len(root_eids) >= 2, root_eids @pytest.mark.parametrize( @@ -354,9 +388,15 @@ def test_sort_key_tie_node_before_line(client, headers, db_exec): 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) + assert idx > 0 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) + node = next((r for r in patch if r["row_type"] in TREE_TYPES + and r["data"]["header"]["expense_item_id"] == child), None) + assert node is not None, "узел дочерней статьи отсутствует в ответе" + line_row = next(r for r in patch if r["row_type"] == "INPUT" and r["data"]["line_id"] == child) + assert node["sort_order"] < line_row["sort_order"] assert_patch_rows_match_sheet(patch, _get_sheet(client, headers, form_id, "AHR")) @@ -368,6 +408,7 @@ def _delta_equals_get(db_exec, sheet, row_id): "SELECT row_type, depth, sort_order, data::text " "FROM v3.v_form_view(4, 'AHR', NULL, 'Support')")} assert delta, "дельта пуста" + assert len({r[2] for r in delta}) == len(delta), "sort_order в дельте не уникален" 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]}") @@ -404,6 +445,7 @@ def test_ahr_delta_includes_sub_sub_item_for_depth4_limit(client, headers, db_ex if not rows: pytest.skip("нет LEAF шаблона лимитов с кодом статьи AHR глубины 4") lt_id, eid = rows[0] + assert lt_id == 13, f"ожидался LEAF 13, найден {lt_id}" _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