128 lines
5.7 KiB
Python
128 lines
5.7 KiB
Python
"""Переиспользуемая проверка контракта ответа на правку ячеек.
|
||
|
||
Каждая строка ответа 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"
|
||
|
||
|
||
# --- AURORA-1850: эталон (фильтр 0050 поверх полного v_form_view) и хелперы SQL ---
|
||
|
||
Row4 = tuple[str, int, int, str] # (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}"
|
||
)
|