test(sheet): эквивалентность ответа на правку и полной сборки — красные тесты (AURORA-1850)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
e3b23e467f
commit
42d47a4c1f
@ -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}"
|
||||
)
|
||||
|
||||
412
api/tests/integration/test_upd_form_cells_subtree.py
Normal file
412
api/tests/integration/test_upd_form_cells_subtree.py
Normal file
@ -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"
|
||||
Loading…
x
Reference in New Issue
Block a user