dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_statement_budget_c.py
"""Statement budgets of slice C's request paths, measured (erp_platform.testing.count_statements) on a group of 2 companies and on one of 10: the
same number of statements, and a pinned ceiling."""
from decimal import Decimal
from conftest import api, seams
from test_slice_c1 import CHANGES, january, year_world
from test_statement_budget import BASE, build, engine_reset, measured
from world import ACC, CTRL, P, ok
D = Decimal
YEAR_BUDGET = {"open_year": 24, "open_year_again": 19} # measured: the same (+4 since audit 1b: the year's result fixed as historical amounts — 3 reads, 1 write)
BUDGET = {"propose_sale": 14, "approve_sale": 19, "eliminate_sale": 24, "propose_stake": 13, "approve_stake": 19} # statements per request (measured: the same)
def sale_paths(gl, engine, extra):
build(gl, extra)
ok(api("POST", "/groups/GRP/rules", ACC, {"account_rules": [{"group_account_id": "G-4900", "valid_from_period": "2020-01", "rate_type": "average"}],
"account_roles": [{"role": "gain_on_disposal", "valid_from_period": "2020-01", "group_account_id": "G-4900"},
{"role": "group_retained_earnings", "valid_from_period": "2020-01", "group_account_id": "G-3200"}]}))
gl.rates[("EUR", "spot", "2026-10-31")] = D("0.880000")
gl.rates[("EUR", "spot", "2026-10-01")] = D("0.870000")
gl.set_rows("DE01", "2026-09", [("G-1100", "1050000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00")])
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
out = {}
stake = {"group_id": "GRP", "kind": "ownership_change", "company_code": "DE01", "investor_company_code": "UK01", "ownership_pct": "90",
"effective_date": "2026-10-01", "consideration": "120000.00"}
out["propose_stake"], made = measured(engine, "POST", CHANGES, ACC, stake)
out["approve_stake"], _ = measured(engine, "POST", f"{CHANGES}/{made['change_id']}/approve-structure-change", CTRL)
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
sale = {"group_id": "GRP", "kind": "divestiture", "company_code": "DE01", "investor_company_code": "UK01", "ownership_pct": "90",
"effective_date": "2026-10-31", "proceeds": "1200000.00"}
out["propose_sale"], made = measured(engine, "POST", CHANGES, ACC, sale)
out["approve_sale"], _ = measured(engine, "POST", f"{CHANGES}/{made['change_id']}/approve-structure-change", CTRL)
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
gl.recs[("UK01", "DE01")] = None
out["eliminate_sale"], _ = measured(engine, "POST", f"{BASE}/run-eliminations", ACC)
return out
def test_ownership_change_and_sale_statements_are_flat_from_2_to_10_companies_and_pinned(gl, engine):
small = sale_paths(gl, engine, 1) # one subsidiary stays after the sale: 3 companies against 10
engine_reset(engine)
gl2 = type(gl)()
seams.ports.gl = gl2
big = sale_paths(gl2, engine, 8)
print("statements", small, big)
assert big == small, (small, big)
assert all(small[k] <= BUDGET[k] for k in small), (small, BUDGET)
def year_paths(gl, engine, extra):
year_world(gl, extra)
january(gl)
out = {}
out["open_year"], _ = measured(engine, "POST", "/groups/GRP/group-periods", ACC, {"period_id": "2027-01"})
out["open_year_again"], _ = measured(engine, "POST", "/groups/GRP/group-periods", ACC, {"period_id": "2027-01"})
return out
def test_opening_a_year_derives_the_carry_forward_in_a_flat_number_of_statements(gl, engine):
small = year_paths(gl, engine, 0)
engine_reset(engine)
gl2 = type(gl)()
seams.ports.gl = gl2
big = year_paths(gl2, engine, 8)
print("statements", small, big)
assert big == small, (small, big)
assert all(small[k] <= YEAR_BUDGET[k] for k in small), (small, YEAR_BUDGET)
# ── slice C2: materialise, the owner's controls, re-collecting and the migration-run sign-off ────────────────────────────────
C2_BUDGET = {"materialise": 15, "set_control": 3, "setting_history": 2, "recollect": 18, "load": 6, "reconcile": 15, "explain": 6, "sign": 6} # measured: the same
def c2_paths(gl, engine, extra):
import json
from test_slice_b2 import CTRL2, publish_versions
from test_slice_c2 import DOORS, RUNS, SYSTEM, carried_history_rows
from world import OWNER, state
from brick_consolidation.domain import consolidation as C
build(gl, extra)
publish_versions(gl)
for step in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step}", ACC, {}))
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
gl.recs[("UK01", "DE01")] = None
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
out = {}
out["materialise"], _ = measured(engine, "POST", f"{BASE}/materialise-statements", SYSTEM)
out["set_control"], _ = measured(engine, "POST", "/controls/allow_open_period_collect", OWNER, {"value": True})
out["setting_history"], _ = measured(engine, "GET", "/settings/allow_open_period_collect/history", ACC)
ok(api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "r"}))
rows = [(r["group_account_id"], r["ytd_amount"], r["period_amount"], r["partner_company"]) for r in gl.figures[("DE01", P)]["rows"]]
gl.set_rows("DE01", P, rows + [("G-6500", "5.00", "5.00"), ("G-2000", "-5.00", "-5.00")])
out["recollect"], _ = measured(engine, "POST", f"{BASE}/collect-group-data", ACC)
carried_history_rows()
members = {c: str(m.ownership) for c, m in sorted(C.members_at(state()[0], P).items())}
odoo = json.dumps([{"check": "consolidated_balance", "odoo": {P: {"G-1100": "922200.00"}}}, {"check": "structure", "odoo": {P: members}},
{"check": "journals", "odoo": {P: {"count": 1, "total": "50.00"}}}, {"check": "proportional", "odoo": {"DE01": "80"}}])
out["load"], _ = measured(engine, "POST", RUNS, SYSTEM, {"cons_migration_run_id": "odoo:X:1", "doors_json": DOORS, "checks_json": odoo})
out["reconcile"], _ = measured(engine, "POST", f"{RUNS}/odoo:X:1/reconcile-cons-migration", SYSTEM)
out["explain"], _ = measured(engine, "POST", f"{RUNS}/odoo:X:1/explain-cons-difference", ACC, {"reason": "decided at the door"})
out["sign"], _ = measured(engine, "POST", f"{RUNS}/odoo:X:1/sign-cons-reconciliation", OWNER)
return out
def test_slice_c2_request_paths_are_flat_from_2_to_10_companies_and_pinned(gl, engine):
small = c2_paths(gl, engine, 0)
engine_reset(engine)
gl2 = type(gl)()
seams.ports.gl = gl2
big = c2_paths(gl2, engine, 8)
print("statements", small, big)
assert big == small, (small, big)
assert all(small[k] <= C2_BUDGET[k] for k in small), (small, C2_BUDGET)