dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_real_gl.py
"""Consolidation against the REAL gl brick, composed in-process (seam.register_local) on gl's own bucket: collect -> translate -> eliminate for a small
group (UK01 GBP, 80% of DE01 EUR). gl is seeded only through its own routes. Wherever the real answer differs from FakeGl's the SPEC decides:
FakeGl is fixed when it is wrong, gl is NEVER edited (a contradiction of gl's own spec is a defect, recorded in the report), and consolidation is
fixed when it misreads a spec-true answer."""
from decimal import Decimal
import pytest
import real_gl as R
from conftest import api, seams
from world import ACC, CTRL, RULES, ok
D = Decimal
P = "2026-10"
BASE = f"/groups/GRP/group-periods/{P}"
ALL_GROUP_CODES = sorted({a["group_account_id"] for a in RULES["account_rules"]})
@pytest.fixture
def real(tmp_path):
with R.mounted(tmp_path):
seams.ports.gl = seams.RealGl()
yield
def seed(skip_average=None, ic_eur="1000.00"):
ids = R.seed_books(ALL_GROUP_CODES)
# capital in December 2025; UK01 pays 900,000.00 EUR x 0.86 for 80% of DE01 on 2026-01-01 (the first month of the group's year)
R.post("CAP-DE", "2025-12-15", "DE01", "EUR", [("1200", {"debit": "1000000.00"}), ("3000", {"credit": "1000000.00"})])
R.post("CAP-UK", "2025-12-15", "UK01", "GBP", [("1200", {"debit": "1500000.00"}), ("3000", {"credit": "1500000.00"})])
R.post("INV-UK", "2026-01-01", "UK01", "GBP", [("1600", {"debit": "774000.00", "partner_company": "DE01"}), ("1200", {"credit": "774000.00"})])
# October: UK01 sells to DE01 for 870.00 GBP (DE01 books 1,000.00 EUR)
R.post("IC-UK", "2026-10-10", "UK01", "GBP", [("1300", {"debit": "870.00", "partner_company": "DE01"}),
("4100", {"credit": "870.00", "partner_company": "DE01"})])
R.post("IC-DE", "2026-10-10", "DE01", "EUR", [("5100", {"debit": ic_eur, "partner_company": "UK01"}),
("2310", {"credit": ic_eur, "partner_company": "UK01"})])
R.match_off()
for co in ("UK01", "DE01"):
R.close_through(P, co)
R.rate("EUR", "GBP", "0.860000", "2026-01-01", "spot")
for m in ("01-31", "02-28", "03-31", "04-30", "05-31", "06-30", "07-31", "08-31", "09-30", "10-31"):
if m == skip_average:
continue
R.rate("EUR", "GBP", "0.860000", f"2026-{m}", "average")
R.rate("EUR", "GBP", "0.870000", "2026-10-31", "closing")
return ids
def build_group(ids):
ok(api("POST", "/groups", ACC, {"group_id": "GRP", "name": "Group", "group_currency": "GBP", "ledger_id": "LEAD", "group_chart_id": "GROUP",
"top_company_code": "UK01", "fiscal_year_end_month": "12"}))
ok(api("POST", "/groups/GRP/rules", ACC, R.real_rules(RULES, ids)))
change = ok(api("POST", "/structure-changes", ACC, {"group_id": "GRP", "kind": "acquisition", "company_code": "DE01", "investor_company_code": "UK01",
"ownership_pct": "80", "method": "full", "effective_date": "2026-01-01", "consideration": "900000.00"}))
ok(api("POST", f"/structure-changes/{change['change_id']}/approve-structure-change", CTRL, {}))
ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
def run_to(step, who=ACC):
return api("POST", f"{BASE}/{step}", who, {})
def entries():
from world import db
return {e["document_type"]: e for e in db(lambda r: r.data.entries("GRP", P, 1))}
def lines(e, ids):
back = {v: k for k, v in ids.items()}
return sorted((back.get(ln["group_account_id"], ln["group_account_id"]), ln["company_code"], ln["partner_company"], str(ln["amount"])) for ln in e["lines"])
def test_collect_translate_eliminate_against_the_real_gl_to_the_penny(real):
ids = seed()
build_group(ids)
out = ok(run_to("collect-group-data"))
assert out["result"]["collected"] == ["DE01", "UK01"]
from world import db
# what is stored IS what gl answered: its rows by its own group account ids, its digest, the period closed, the functional currency
for co, ccy in (("DE01", "EUR"), ("UK01", "GBP")):
answer = seams.ports.gl.balances(co, "LEAD", P)
got = {(r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"]): (r["ytd_local"], r["period_local"])
for r in db(lambda r: r.data.reported("GRP", [P], [co]))}
want = {(r["group_account_id"], r["partner_company"], r["movement_type"], r["segment"]): (D(r["ytd_amount"]), D(r["period_amount"])) for r in answer["rows"]}
assert got == want and answer["unmapped"] == [] and answer["period_status"] == "closed" and answer["functional_currency"] == ccy
[c] = [c for c in db(lambda r: r.data.collections("GRP", P)) if c["company_code"] == co]
assert (c["gl_digest"], c["period_status"], c["local_currency"]) == (answer["digest"], "closed", ccy)
assert sum(v[0] for v in got.values()) == 0
ok(run_to("translate-currencies"))
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
ok(run_to("run-eliminations"))
e = entries()
assert lines(e["translation"], ids) == [("G-3900", "DE01", "", "-7992.00"), ("G-3950", "DE01", "", "-1998.00")] # 9,990.00: 80% / 20%
assert lines(e["ic_balance"], ids) == [("G-1300", "UK01", "DE01", "-870.00"), ("G-2300", "DE01", "UK01", "870.00")] # 1,000.00 EUR x 0.87
assert lines(e["ic_pl"], ids) == [("G-4100", "UK01", "DE01", "870.00"), ("G-5100", "DE01", "UK01", "-860.00"), ("G-6900", "UK01", "DE01", "-10.00")]
inv = lines(e["investment"], ids) # IAS 21.47: goodwill 100,000.00 EUR is shown at the closing 0.87 (87,000.00), not the acquisition rate (86,000.00)
assert ("G-1700", "DE01", "", "87000.00") in inv and ("G-1600", "UK01", "DE01", "-774000.00") in inv
assert ("G-3900", "DE01", "", "-1000.00") in inv # the 1,000.00 gain on the goodwill goes to the translation reserve, not to profit
assert all(sum(ln["amount"] for ln in x["lines"]) == 0 for k, x in e.items() if k != "translation") # every elimination nets to zero, exactly
def test_a_missing_months_average_is_refused_not_translated_with_the_month_before(real):
ids = seed(skip_average="10-31") # no average is dated October's last day
build_group(ids)
ok(run_to("collect-group-data"))
older = R.must(R.gl("GET", "/fx-rates/EUR", R.ACCOUNTANT, {"on": "2026-10-31", "type": "average", "to": "GBP"}))
assert older["rate_date"] == "2026-09-30" # `on` answers the newest rate on or before the day: September's
s, miss = R.gl("GET", "/fx-rates/EUR", R.ACCOUNTANT, {"month": "2026-10", "type": "average", "to": "GBP"})
assert s == 404 and miss["gate"] == "fx_rate" # the month asks for exactly October's: never the nearest
status, text = run_to("translate-currencies") # so consolidation asks by month, and refuses rather than use September's
assert status == 409 and "average rate EUR" in str(text) and "2026-10" in str(text) and "never used" in str(text), text
def test_no_match_run_in_gl_is_a_404_and_the_pair_out_of_tolerance_is_blocked_with_gls_fix(real):
ids = seed(ic_eur="1100.00") # DE01 owes 1,100.00 EUR = 957.00 GBP against UK01's 870.00: 87.00 apart (tolerance 5.00)
build_group(ids)
for step_ in ("collect-group-data", "translate-currencies"):
ok(run_to(step_))
assert R.gl("GET", "/intercompany/reconciliation", R.CONSOLIDATION, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"})[0] == 404
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
status, text = run_to("run-eliminations")
assert status == 409 and "ic_within_tolerance" in str(text) and "run match_intercompany in gl" in str(text), text
# ── the spec decides: what real gl answers against what its signed spec says ───────────────────────────────────────────────
def test_the_real_answers_carry_every_field_the_spec_lists(real):
import yaml
spec = yaml.safe_load(open(f"{__import__('contract_checks').METHOD_DIR}/modules/gl/module.yaml")) # read-only: erp-method is never edited
assert spec # (the shapes below are the seams' response lines)
ids = seed()
build_group(ids)
gl = seams.ports.gl
b = gl.balances("DE01", "LEAD", P)
assert set(b) >= {"company_code", "ledger_id", "period_id", "fiscal_year", "period_status", "functional_currency", "group_chart_id", "digest", "rows", "unmapped"}
assert set(b["rows"][0]) >= {"group_account_id", "partner_company", "movement_type", "segment", "ytd_amount", "period_amount"}
assert set(R.must(R.gl("GET", "/fx-rates/EUR", R.ACCOUNTANT, {"on": "2026-10-31", "type": "closing", "to": "GBP"}))) >= {
"currency", "to", "type", "on", "rate_date", "rate", "source"}
assert set(gl.period_status("UK01", __import__("datetime").date(2026, 10, 31))) >= {"date", "company_code", "period_id", "open", "status", "reason", "fix"}
[dict_company] = [c for c in gl.companies() if c["company_code"] == "DE01"]
assert set(dict_company) >= {"company_code", "name", "country", "functional_currency", "chart_id", "bp_id", "financial_year_end_month", "ledgers"}
assert {a["code"] for a in gl.chart_accounts("GROUP")} == set(ALL_GROUP_CODES) and set(gl.chart_accounts("GROUP")[0]) >= {"account_id", "code", "name", "type", "subtype"}
assert gl.migration_map("ODOO", "res.company", ["1", "2"]) == {} # nothing migrated: omitted, never an error
with pytest.raises(seams.Missing) as no_version:
gl.statement_version("NOPE")
assert no_version.value.status == 404
def test_gl_defect_d1_companies_is_a_bare_list_as_the_spec_says(real):
seed()
body = R.must(R.gl("GET", "/companies", R.CONSOLIDATION_CALLER))
assert isinstance(body, list)
def test_gl_defect_d2_chart_accounts_is_a_bare_list_as_the_spec_says(real):
seed()
assert isinstance(R.must(R.gl("GET", "/charts/GROUP/accounts", R.CONSOLIDATION_CALLER)), list)
def test_cons_and_the_real_gl_compose_with_no_collision(tmp_path):
"""tests/tier1.compose.yaml without business-partner (not installed here): no table or tool collision, versions as pinned."""
import yaml
from erp_platform import compose
spec = yaml.safe_load(open(__import__("pathlib").Path(__file__).parent / "tier1.compose.yaml"))
spec["bricks"] = [b for b in spec["bricks"] if b["name"] != "business-partner"]
path = tmp_path / "cons-gl.compose.yaml"
path.write_text(yaml.safe_dump(spec))
assert compose.check(compose.load(path))["problems"] == []