test(form4): суммы ROOT/GROUP/ITEM листа CAP против v3.jsonb_sum и Python, фикстура db_session
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Andrey Epifancev
parent
2a287480b6
commit
55ba4e7e96
@ -13,8 +13,15 @@ from src.main import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client():
|
||||
def db_session():
|
||||
"""Сессия БД теста: её же использует client, всё откатывается в конце теста."""
|
||||
db_ref = SessionLocal()
|
||||
yield db_ref
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(db_session):
|
||||
db_ref = db_session
|
||||
loop_ref = None
|
||||
|
||||
async def _get_db():
|
||||
|
||||
344
api/tests/integration/test_form4_sheet_sums.py
Normal file
344
api/tests/integration/test_form4_sheet_sums.py
Normal file
@ -0,0 +1,344 @@
|
||||
"""AURORA-1846: суммы ROOT/GROUP/ITEM на листе FORM_4/CAP.
|
||||
|
||||
Два эталона:
|
||||
1. старый рекурсивный агрегат v3.jsonb_sum поверх INPUT-строк (точное совпадение,
|
||||
включая масштаб чисел и правило «текст сохраняется при одном вхождении»);
|
||||
2. независимая сумма числовых листьев на Python.
|
||||
"""
|
||||
import json
|
||||
from decimal import Decimal
|
||||
from itertools import count
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
|
||||
SHEET = "CAP"
|
||||
CAP_ITEM = 525 # статья CAP depth=2
|
||||
BLOCKS = [
|
||||
"plan", "seq_dfip", "approved", "contract_summary", "allocation", "reserve",
|
||||
"collegial", "ckk", "contract", "booking", "q1", "q2", "q3", "q4", "totals",
|
||||
]
|
||||
_names = count(1)
|
||||
|
||||
|
||||
def _loads(s):
|
||||
return json.loads(s, parse_float=Decimal)
|
||||
|
||||
|
||||
def canon(x):
|
||||
"""Decimal -> str (с масштабом), чтобы 3.50 и 3.5 считались разными."""
|
||||
if isinstance(x, Decimal):
|
||||
return str(x)
|
||||
if isinstance(x, dict):
|
||||
return {k: canon(v) for k, v in x.items()}
|
||||
if isinstance(x, list):
|
||||
return [canon(v) for v in x]
|
||||
return x
|
||||
|
||||
|
||||
def canon_num(x):
|
||||
"""Числа -> нормализованный Decimal: API отдаёт float, масштаб теряется."""
|
||||
if isinstance(x, bool):
|
||||
return x
|
||||
if isinstance(x, (int, float, Decimal)):
|
||||
return Decimal(str(x)).normalize()
|
||||
if isinstance(x, dict):
|
||||
return {k: canon_num(v) for k, v in x.items()}
|
||||
if isinstance(x, list):
|
||||
return [canon_num(v) for v in x]
|
||||
return x
|
||||
|
||||
|
||||
def leaves(block):
|
||||
"""(путь, значение) для объекта глубиной до 2."""
|
||||
if not isinstance(block, dict):
|
||||
return
|
||||
for k1, v1 in block.items():
|
||||
if isinstance(v1, dict):
|
||||
for k2, v2 in v1.items():
|
||||
yield (k1, k2), v2
|
||||
else:
|
||||
yield (k1,), v1
|
||||
|
||||
|
||||
def get_path(block, path):
|
||||
for k in path:
|
||||
if not isinstance(block, dict) or k not in block:
|
||||
return None
|
||||
block = block[k]
|
||||
return block
|
||||
|
||||
|
||||
class F4:
|
||||
def __init__(self, client, db_session, headers):
|
||||
self.client, self.db, self.h = client, db_session, headers
|
||||
resp = client.post(
|
||||
"/api/v1/form/",
|
||||
json={"form_type_code": "FORM_4", "year": 2026, "org_unit_ids": [2]},
|
||||
headers=headers,
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
self.form_id = resp.json()["result"][0]["id"]
|
||||
|
||||
def sql(self, query, **params):
|
||||
async def _go():
|
||||
r = await self.db.execute(text(query), params)
|
||||
return r.fetchall() if r.returns_rows else []
|
||||
|
||||
return self.client.portal.call(_go)
|
||||
|
||||
def program(self, name=None):
|
||||
name = name or f"prog{next(_names)}"
|
||||
return self.sql(
|
||||
"SELECT project_id FROM v3.add_form4_program(:n, :f, 'CAP') LIMIT 1",
|
||||
n=name, f=self.form_id,
|
||||
)[0][0]
|
||||
|
||||
def project(self, program_id, name=None):
|
||||
name = name or f"prj{next(_names)}"
|
||||
return self.sql(
|
||||
"SELECT project_id FROM v3.add_form4_project(:n, :p) LIMIT 1",
|
||||
n=name, p=program_id,
|
||||
)[0][0]
|
||||
|
||||
def line(self, project_id):
|
||||
before = {r[0] for r in self.sql("SELECT id FROM v3.budget_line WHERE budget_form_id=:f", f=self.form_id)}
|
||||
resp = self.client.post(
|
||||
f"/api/v1/form/{self.form_id}/sheet/{SHEET}/line",
|
||||
json={"expense_item_id": CAP_ITEM, "project_id": project_id, "name": f"line{next(_names)}"},
|
||||
headers=self.h,
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
new = {r[0] for r in self.sql("SELECT id FROM v3.budget_line WHERE budget_form_id=:f", f=self.form_id)} - before
|
||||
assert len(new) == 1
|
||||
return new.pop()
|
||||
|
||||
def cell(self, line_id, column, value):
|
||||
self.sql(
|
||||
"SELECT v3._apply_form_cell(:l, :c, CAST(:v AS jsonb), 'CAP', :f)",
|
||||
l=line_id, c=column, v=json.dumps(value), f=self.form_id,
|
||||
)
|
||||
|
||||
def cells(self, line_id, values):
|
||||
for col, val in values.items():
|
||||
self.cell(line_id, col, val)
|
||||
|
||||
def sheet(self, sections=None):
|
||||
params = {"sections": ",".join(sections)} if sections else {}
|
||||
resp = self.client.get(f"/api/v1/form/{self.form_id}/sheet/{SHEET}", params=params, headers=self.h)
|
||||
assert resp.status_code == 200, resp.text
|
||||
return _loads(resp.text)["result"]
|
||||
|
||||
def reference(self, sections=None):
|
||||
"""Эталон 1: старый v3.jsonb_sum по INPUT-строкам, по уровням."""
|
||||
sums = ", ".join(f"'{b}', v3.jsonb_sum(data->'{b}')" for b in BLOCKS)
|
||||
q = f"""
|
||||
SELECT lvl, gid, jsonb_build_object({sums})::text
|
||||
FROM (
|
||||
SELECT 'P' AS lvl, (data->'header'->>'program_id')::int AS gid, data
|
||||
FROM v3.v_form4_sheet_jsonb(:f, 'CAP', CAST(:s AS text[])) WHERE row_type = 'INPUT'
|
||||
AND data->'header'->>'program_id' IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT 'J', (data->'header'->>'project_id')::int, data
|
||||
FROM v3.v_form4_sheet_jsonb(:f, 'CAP', CAST(:s AS text[])) WHERE row_type = 'INPUT'
|
||||
AND data->'header'->>'project_id' IS NOT NULL
|
||||
UNION ALL
|
||||
SELECT 'R', 0, data
|
||||
FROM v3.v_form4_sheet_jsonb(:f, 'CAP', CAST(:s AS text[])) WHERE row_type = 'INPUT'
|
||||
) x
|
||||
GROUP BY lvl, gid
|
||||
"""
|
||||
rows = self.sql(q, f=self.form_id, s=sections)
|
||||
return {(lvl, gid): _loads(js) for lvl, gid, js in rows}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def f4(client, db_session, admin_tokens, auth_headers):
|
||||
return F4(client, db_session, auth_headers(admin_tokens))
|
||||
|
||||
|
||||
def _split(rows):
|
||||
root = [r for r in rows if r["row_type"] == "ROOT"]
|
||||
assert len(root) == 1
|
||||
groups = {r["data"]["header"]["program_id"]: r["data"] for r in rows if r["row_type"] == "GROUP"}
|
||||
items = {r["data"]["header"]["project_id"]: r["data"] for r in rows if r["row_type"] == "ITEM"}
|
||||
inputs = [r["data"] for r in rows if r["row_type"] == "INPUT"]
|
||||
return root[0]["data"], groups, items, inputs
|
||||
|
||||
|
||||
def check_sums(f4, sections=None):
|
||||
"""Сверка листа с обоими эталонами. Возвращает (root, groups, items, inputs)."""
|
||||
rows = f4.sheet(sections)
|
||||
root, groups, items, inputs = _split(rows)
|
||||
ref = f4.reference(sections)
|
||||
|
||||
# Эталон 1a: ответ API совпадает с v3.jsonb_sum (числа сравниваются по значению)
|
||||
assert canon_num({b: root[b] for b in BLOCKS}) == canon_num(ref[("R", 0)])
|
||||
for pid, data in groups.items():
|
||||
if ("P", pid) in ref:
|
||||
assert canon_num({b: data[b] for b in BLOCKS}) == canon_num(ref[("P", pid)]), f"program {pid}"
|
||||
for jid, data in items.items():
|
||||
if ("J", jid) in ref:
|
||||
assert canon_num({b: data[b] for b in BLOCKS}) == canon_num(ref[("J", jid)]), f"project {jid}"
|
||||
|
||||
# Эталон 1b: прямой вызов функции совпадает с v3.jsonb_sum точно, включая масштаб чисел
|
||||
direct = f4.sql(
|
||||
"SELECT row_type, data::text FROM v3.v_form4_sheet_jsonb(:f, 'CAP', CAST(:s AS text[]))",
|
||||
f=f4.form_id, s=sections,
|
||||
)
|
||||
d_root, d_groups, d_items, _ = _split([{"row_type": t, "data": _loads(js)} for t, js in direct])
|
||||
assert canon({b: d_root[b] for b in BLOCKS}) == canon(ref[("R", 0)])
|
||||
for pid, data in d_groups.items():
|
||||
if ("P", pid) in ref:
|
||||
assert canon({b: data[b] for b in BLOCKS}) == canon(ref[("P", pid)]), f"program {pid}"
|
||||
for jid, data in d_items.items():
|
||||
if ("J", jid) in ref:
|
||||
assert canon({b: data[b] for b in BLOCKS}) == canon(ref[("J", jid)]), f"project {jid}"
|
||||
|
||||
# Эталон 2: сумма числовых листьев на Python
|
||||
def expect(level_inputs, agg):
|
||||
for b in BLOCKS:
|
||||
paths = {p for inp in level_inputs for p, _ in leaves(inp[b])}
|
||||
for path in paths:
|
||||
nums = [
|
||||
Decimal(str(v))
|
||||
for inp in level_inputs
|
||||
for p, v in leaves(inp[b])
|
||||
if p == path and isinstance(v, (int, Decimal)) and not isinstance(v, bool)
|
||||
]
|
||||
if nums:
|
||||
assert Decimal(str(get_path(agg[b], path))) == sum(nums), (b, path)
|
||||
|
||||
expect(inputs, root)
|
||||
for pid, data in groups.items():
|
||||
lvl_inputs = [i for i in inputs if i["header"]["program_id"] == pid]
|
||||
if lvl_inputs:
|
||||
expect(lvl_inputs, data)
|
||||
for jid, data in items.items():
|
||||
lvl_inputs = [i for i in inputs if i["header"]["project_id"] == jid]
|
||||
if lvl_inputs:
|
||||
expect(lvl_inputs, data)
|
||||
return root, groups, items, inputs
|
||||
|
||||
|
||||
def test_program_with_two_projects(f4):
|
||||
prog = f4.program()
|
||||
p1, p2 = f4.project(prog), f4.project(prog)
|
||||
for prj, vals in ((p1, (10, 20)), (p1, (5, 7)), (p2, (100, 200)), (p2, (1, 2))):
|
||||
ln = f4.line(prj)
|
||||
f4.cells(ln, {"plan.q1": vals[0], "plan.q2": vals[1], "reserve.q1": vals[0]})
|
||||
root, groups, items, _ = check_sums(f4)
|
||||
assert Decimal(str(items[p1]["plan"]["q1"])) == 15
|
||||
assert Decimal(str(items[p2]["plan"]["q2"])) == 202
|
||||
assert Decimal(str(groups[prog]["plan"]["q1"])) == 116
|
||||
assert Decimal(str(root["plan"]["q1"])) == 116
|
||||
|
||||
|
||||
def test_lines_directly_on_program(f4):
|
||||
prog = f4.program()
|
||||
prj = f4.project(prog)
|
||||
direct = f4.line(prog)
|
||||
in_prj = f4.line(prj)
|
||||
f4.cell(direct, "plan.q1", 40)
|
||||
f4.cell(in_prj, "plan.q1", 2)
|
||||
root, groups, items, _ = check_sums(f4)
|
||||
assert Decimal(str(groups[prog]["plan"]["q1"])) == 42
|
||||
assert Decimal(str(items[prj]["plan"]["q1"])) == 2
|
||||
assert Decimal(str(root["plan"]["q1"])) == 42
|
||||
|
||||
|
||||
def test_single_line_project_keeps_text_two_lines_null(f4):
|
||||
prog = f4.program()
|
||||
single, pair = f4.project(prog), f4.project(prog)
|
||||
ln = f4.line(single)
|
||||
f4.cells(ln, {"contract.counterparty": "ООО Один", "q1.adj_comment": "комментарий", "plan.q1": 3})
|
||||
for txt in ("ООО А", "ООО Б"):
|
||||
f4.cells(f4.line(pair), {"contract.counterparty": txt, "q1.adj_comment": txt, "plan.q1": 1})
|
||||
_, groups, items, _ = check_sums(f4)
|
||||
assert items[single]["contract"]["counterparty"] == "ООО Один"
|
||||
assert items[single]["q1"]["adj_comment"] == "комментарий"
|
||||
assert items[pair]["contract"]["counterparty"] is None
|
||||
assert items[pair]["q1"]["adj_comment"] is None
|
||||
assert groups[prog]["contract"]["counterparty"] is None
|
||||
|
||||
|
||||
def test_single_line_program_keeps_text(f4):
|
||||
prog = f4.program()
|
||||
f4.cells(f4.line(prog), {"contract.counterparty": "ООО Одна", "plan.q1": 1})
|
||||
_, groups, _, _ = check_sums(f4)
|
||||
assert groups[prog]["contract"]["counterparty"] == "ООО Одна"
|
||||
|
||||
|
||||
def test_empty_program_and_project(f4):
|
||||
filled = f4.program()
|
||||
f4.cell(f4.line(filled), "plan.q1", 9)
|
||||
empty_prog = f4.program()
|
||||
empty_prj = f4.project(empty_prog)
|
||||
_, groups, items, _ = check_sums(f4)
|
||||
zero_plan = {"q1": 0, "q2": 0, "q3": 0, "q4": 0, "year": 0, "comment": None}
|
||||
assert canon(groups[empty_prog]["plan"]) == canon(zero_plan)
|
||||
assert canon(items[empty_prj]["plan"]) == canon(zero_plan)
|
||||
|
||||
|
||||
def test_decimals_negative_and_null(f4):
|
||||
prog = f4.program()
|
||||
prj = f4.project(prog)
|
||||
other = f4.project(prog)
|
||||
a, b, c, d = f4.line(prj), f4.line(prj), f4.line(prj), f4.line(other)
|
||||
f4.cells(a, {"plan.q1": "1.50", "plan.q2": -3, "plan.q3": "0.125", "reserve.q1": 7})
|
||||
f4.cells(b, {"plan.q1": 2, "plan.q2": "-0.5", "plan.q3": "10.1", "reserve.q1": None})
|
||||
f4.cells(c, {"plan.q1": None, "plan.q2": "2.000", "plan.q3": None})
|
||||
f4.cells(d, {"plan.q1": "-1.5", "plan.q2": 0})
|
||||
_, groups, items, _ = check_sums(f4)
|
||||
assert Decimal(str(items[prj]["plan"]["q1"])) == Decimal("3.5")
|
||||
assert Decimal(str(items[prj]["plan"]["q2"])) == Decimal("-1.5")
|
||||
assert Decimal(str(groups[prog]["plan"]["q1"])) == 2
|
||||
|
||||
|
||||
def test_booking_second_level(f4):
|
||||
prog = f4.program()
|
||||
prj = f4.project(prog)
|
||||
for q1, q2, nx in ((1, "2.5", 4), (10, None, "0.75"), (None, 3, 5)):
|
||||
f4.cells(f4.line(prj), {"booking.y2026.q1": q1, "booking.y2026.q2": q2, "booking.y2027.q1": nx})
|
||||
_, groups, items, _ = check_sums(f4)
|
||||
bk = items[prj]["booking"]
|
||||
assert Decimal(str(bk["y2026"]["q1"])) == 11
|
||||
assert Decimal(str(bk["y2026"]["q2"])) == Decimal("5.5")
|
||||
assert Decimal(str(bk["y2027"]["q1"])) == Decimal("9.75")
|
||||
assert Decimal(str(groups[prog]["booking"]["y2027"]["q1"])) == Decimal("9.75")
|
||||
|
||||
|
||||
def test_sections_filter(f4):
|
||||
prog = f4.program()
|
||||
prj = f4.project(prog)
|
||||
for v in (1, 2):
|
||||
f4.cells(f4.line(prj), {"plan.q1": v, "booking.y2026.q1": v * 10, "reserve.q1": v, "contract.counterparty": "x"})
|
||||
root, groups, items, _ = check_sums(f4, sections=["plan", "booking"])
|
||||
assert Decimal(str(items[prj]["plan"]["q1"])) == 3
|
||||
assert Decimal(str(items[prj]["booking"]["y2026"]["q1"])) == 30
|
||||
for b in BLOCKS:
|
||||
if b not in ("plan", "booking"):
|
||||
assert root[b] is None and groups[prog][b] is None and items[prj][b] is None
|
||||
|
||||
|
||||
def test_cell_patch_returns_updated_ancestor_sums(f4, client):
|
||||
prog = f4.program()
|
||||
prj = f4.project(prog)
|
||||
l1, l2 = f4.line(prj), f4.line(prj)
|
||||
f4.cells(l1, {"plan.q1": 10})
|
||||
f4.cells(l2, {"plan.q1": 5})
|
||||
resp = client.patch(
|
||||
f"/api/v1/form/{f4.form_id}/sheet/{SHEET}/cell",
|
||||
json={"line_id": l1, "column": "plan.q1", "value": 20},
|
||||
headers=f4.h,
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
rows = _loads(resp.text)["result"]
|
||||
by_type = {r["row_type"]: r["data"] for r in rows}
|
||||
assert {"ROOT", "GROUP", "ITEM"} <= set(by_type), set(by_type)
|
||||
for t in ("ROOT", "GROUP", "ITEM"):
|
||||
assert Decimal(str(by_type[t]["plan"]["q1"])) == 25, t
|
||||
# то, что отдал PATCH, совпадает со свежим GET
|
||||
_, groups, items, _ = check_sums(f4)
|
||||
assert Decimal(str(groups[prog]["plan"]["q1"])) == 25
|
||||
assert Decimal(str(items[prj]["plan"]["q1"])) == 25
|
||||
Loading…
x
Reference in New Issue
Block a user