dodil/erp-brick-consolidation

DODIL ERP brick: consolidation

tests/test_statement_budget.py

"""Statement budgets: live DataK3 costs ~70-100 ms a statement, so a request's statements must not grow with the companies. Each path is
measured (erp_platform.testing.count_statements) on a group of 2 companies and on one of 10 and must issue the same number of statements;
the number is pinned."""
from decimal import Decimal

from conftest import Repository, api, md, seams
from erp_platform.testing import count_statements
from test_slice_b2 import CSV, CTRL2, publish_versions, us01_world
from world import ACC, CTRL, UK01_ROWS, P, acquire, group, ok

D = Decimal
BASE = f"/groups/GRP/group-periods/{P}"
BUDGET = {"collect": 19, "translate": 26, "eliminate": 26, "propose_entry": 20, "approve_entry": 23, "reverse_entry": 26, "readiness": 15, "report": 12,
          "report_signed": 9, "sign_off": 36, "reopen": 15, "upload": 14, "accept": 19}      # statements per request (measured 17 24 24 18 21 24 13 10 7 34 13 12 17; v3 sign_off +3: the tax_rate_known readiness gate)


def build(gl, extra):
    group(gl)
    acquire("DE01")
    investments = []
    for i in range(extra):
        code = f"C{i:02d}"
        gl.company(code, "EUR", fy_end=3)
        gl.set_rows(code, "2026-03", [("G-1100", "100.00", "0.00"), ("G-3000", "-100.00", "0.00")])
        gl.set_rows(code, P, [("G-1100", "100.00", "0.00"), ("G-2000", "-90.00", "0.00"), ("G-3000", "-10.00", "0.00")])
        acquire(code, "100", "100.00")
        investments.append(("G-1600", "86.00", "0.00", code))
    capital = [("G-3000", str(D("-742350.00") - 86 * extra), "0.00") if r[0] == "G-3000" else r for r in UK01_ROWS]       # one row per key
    gl.set_rows("UK01", P, capital + investments)
    ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
    publish_versions(gl)


def measured(engine, method, path, headers, body=None):
    with count_statements(engine) as counted:
        out = ok(api(method, path, headers, body or {}))
    return counted.statements, out


def run_all(gl, engine, extra):
    build(gl, extra)
    out = {}
    out["collect"], _ = measured(engine, "POST", f"{BASE}/collect-group-data", ACC)
    out["translate"], _ = measured(engine, "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"], _ = measured(engine, "POST", f"{BASE}/run-eliminations", ACC)
    lines = [{"group_account_id": "G-1700", "company_code": "DE01", "amount": "10.00"}, {"group_account_id": "G-1700", "company_code": "UK01", "amount": "-10.00"}]
    body = {"cons_journal_id": "E1", "group_id": "GRP", "period_id": P, "lines": lines, "memo": "m"}
    out["propose_entry"], made = measured(engine, "POST", "/cons-journals/propose-cons-entry", ACC, body)
    out["approve_entry"], _ = measured(engine, "POST", f"/cons-journals/{made['cons_journal_id']}/approve-cons-entry", CTRL)
    out["reverse_entry"], _ = measured(engine, "POST", f"/cons-journals/{made['cons_journal_id']}/reverse-cons-entry", ACC, {"reason": "r"})
    out["readiness"], ready = measured(engine, "GET", f"{BASE}/readiness", CTRL)
    assert ready["met"], [g for g in ready["gates"] if not g["met"]]
    out["report"], _ = measured(engine, "GET", f"{BASE}/report", ACC, {"fsv_id": "GROUP-IFRS"})
    out["sign_off"], _ = measured(engine, "POST", f"{BASE}/sign-off-group-close", CTRL)
    out["report_signed"], _ = measured(engine, "GET", f"{BASE}/report", ACC, {"fsv_id": "GROUP-IFRS"})
    out["reopen"], _ = measured(engine, "POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "r"})
    return out


def engine_reset(engine):
    """A fresh database file for the second group: drop every row, keep the tables."""
    with engine.begin() as conn:
        for t in reversed(md.sorted_tables):
            conn.execute(t.delete())
    with engine.begin() as conn:
        Repository(conn).seed()


def test_statements_per_request_are_flat_from_2_to_10_companies_and_pinned(gl, engine):
    small = run_all(gl, engine, 0)                 # UK01 and DE01
    engine_reset(engine)
    gl2 = type(gl)()
    seams.ports.gl = gl2
    big = run_all(gl2, engine, 8)                  # UK01, DE01 and 8 more: 10 companies
    print("statements", small, big)
    assert big == small, (small, big)
    assert all(small[k] <= BUDGET[k] for k in small if k in BUDGET), (small, BUDGET)


def uploads(gl, engine, extra):
    """Upload and accept for a company kept outside gl, in a group with `extra` more companies in its structure."""
    from brick_consolidation.application import use_cases
    from brick_consolidation.domain.group import GroupStructure
    us01_world(gl)
    with use_cases.repository() as repo:
        g, v = repo.groups.get("GRP")
        g.group_structures += [GroupStructure(group_id="GRP", company_code=f"X{i:02d}", valid_from_period="2026-04", investor_company_code="UK01",
                                              ownership_pct=D("100"), voting_pct=D("100"), method="full", change_id="elsewhere") for i in range(extra)]
        repo.groups.save(g, v)
    out = {}
    out["upload"], _ = measured(engine, "POST", f"{BASE}/upload-reported-data", ACC, {"company_code": "US01", "currency": "USD", "csv": CSV})
    out["accept"], _ = measured(engine, "POST", f"{BASE}/accept-reported-upload", CTRL, {"company_code": "US01", "currency": "USD"})
    return out


def test_upload_and_accept_are_flat_from_2_to_10_companies_and_pinned(gl, engine):
    small = uploads(gl, engine, 0)
    engine_reset(engine)
    gl2 = type(gl)()
    seams.ports.gl = gl2
    big = uploads(gl2, engine, 8)
    print("statements", small, big)
    assert big == small, (small, big)
    assert all(small[k] <= BUDGET[k] for k in ("upload", "accept")), (small, BUDGET)