test(sheet): усиление красных тестов эквивалентности по ревью Task 2 (AURORA-1850)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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",

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(
+    assert len(sibs) == 2, "нет пары сестёр, отличных от первой/последней"
-        (lf for lf in leaves if lf[2][0] not in (first[2][0], last[2][0]) and lf not in chosen + sibs),
+    n_roots = len({lf[2][0] for lf in leaves})
-        None,
+    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