dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_d.py
"""Slice D: CON-30..33 (what consolidation reads from gl) and CON-34 (the Odoo account_consolidation loader), through the doors with gl as the fake."""
import json
from decimal import Decimal
from conftest import api, seams
from fake_odoo import FakeOdoo, gl_signed
from test_slice_c2 import RUNS, SYSTEM, step
from world import ACC, OWNER, P, db, ok, period_row, state, world
D = Decimal
BASE = f"/groups/GRP/group-periods/{P}"
# ── CON-30 ─────────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_30_the_rows_the_period_status_the_currency_the_unmapped_and_the_digest_come_back_and_are_stored_as_reported_data(gl):
world(gl)
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
assert ("balances", "UK01") in gl.calls and ("balances", "DE01") in gl.calls
answer = gl.balances("UK01", "IFRS", P)
assert set(answer) >= {"company_code", "ledger_id", "period_id", "fiscal_year", "period_status", "functional_currency", "group_chart_id", "digest", "rows", "unmapped"}
assert answer["period_status"] == "closed" and answer["functional_currency"] == "GBP" and answer["unmapped"] == []
stored = {(r["company_code"], r["group_account_id"], r["partner_company"]): r for r in db(lambda r: r.data.reported("GRP", [P], ["UK01"]))}
got = stored[("UK01", "G-1300", "DE01")]
assert (got["ytd_local"], got["local_currency"], got["source"]) == (D("870.00"), "GBP", "gl")
coll = {c["company_code"]: c for c in db(lambda r: r.data.collections("GRP", P))} # gl's digest and the period status, kept per company and run
assert (coll["UK01"]["gl_digest"], coll["UK01"]["period_status"], coll["UK01"]["local_currency"]) == (answer["digest"], "closed", "GBP")
# consolidation never reads gl's tables: it owns only cons_ tables, and gl's rows reach it by the route
from brick_consolidation.infrastructure.tables import md
assert not [t for t in md.tables if t.startswith("gl_")]
def test_con_30_accounts_gl_could_not_map_are_listed_and_a_missing_ledger_or_company_is_gls_refusal_in_gls_words(gl):
world(gl)
gl.set_rows("DE01", P, [r for r in __import__("world").DE01_ROWS], unmapped=[{"account_id": "L-77", "code": "7700", "ytd_amount": "5.00"}])
status, text = api("POST", f"{BASE}/collect-group-data", ACC, {})
assert status == 409 and "7700" in str(text) and period_row().status == "open"
del gl.figures[("DE01", P)] # gl: 422 ledger_assigned
status, text = api("POST", f"{BASE}/collect-group-data", ACC, {})
assert status == 409 and "ledger_assigned" in str(text)
del gl.companies_["DE01"] # gl: 404, the company is not visible
status, text = api("POST", f"{BASE}/collect-group-data", ACC, {})
assert status == 409 and "not visible" in str(text) and period_row().status == "open"
# ── CON-31 ─────────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_31_an_accepted_break_goes_to_ic_difference_an_unaccepted_one_blocks_and_no_match_run_is_gls_404_with_its_fix(gl):
from test_slice_a2 import entries, ic_ready, lines
from world import UK01_ROWS
world(gl)
gl.set_rows("UK01", P, [("G-1300", "920.00", "0.00", "DE01") if r[0] == "G-1300" else ("G-3000", "-742400.00", "0.00") if r[0] == "G-3000" else r
for r in UK01_ROWS])
for step_ in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step_}", ACC, {}))
ic_ready(gl)
gl.recs.clear() # 52.00 apart and gl has no match run for the pair
status, text = api("POST", f"{BASE}/run-eliminations", ACC, {})
assert status == 409 and "ic_within_tolerance" in str(text) and "run match_intercompany in gl" in str(text), text
assert period_row().status == "translated" and entries("ic_balance") == []
ic_ready(gl, accepted=False) # a match run with an unaccepted break above tolerance: blocked
gl.recs[("UK01", "DE01")]["breaks"][0].update({"ours": "920.00", "difference": "52.00", "within_tolerance": False})
status, text = api("POST", f"{BASE}/run-eliminations", ACC, {})
assert status == 409 and "ic_within_tolerance" in str(text) and "INV-77" in str(text) and entries("ic_balance") == []
gl.recs[("UK01", "DE01")]["breaks"][0].update({"difference": "52.00", "within_tolerance": True, "accepted": True}) # accepted: eliminated
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[entry] = entries("ic_balance", "posted")
assert lines(entry)[("G-1990", "UK01", "DE01")] == D("52.00") and ("reconciliation", "UK01", "DE01") in gl.calls # to ic_difference
# ── CON-32 ─────────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_32_the_rates_in_force_come_back_and_a_404_names_currency_type_and_date_and_blocks_translation(gl):
world(gl)
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
closing = gl.rates.pop(("EUR", "closing", "2026-10-31"))
status, text = api("POST", f"{BASE}/translate-currencies", ACC, {})
assert status == 409 and "EUR" in str(text) and "closing" in str(text) and "2026-10-31" in str(text) and "add it in gl" in str(text), text
assert period_row().status == "collected"
gl.rates[("EUR", "closing", "2026-10-31")] = closing
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
fx = [c for c in gl.calls if c[0] == "fx"]
assert ("fx", "EUR", "closing", "2026-10-31") in fx
for month_end in ("2026-04-30", "2026-05-31", "2026-06-30", "2026-07-31", "2026-08-31", "2026-09-30", "2026-10-31"):
assert ("fx", "EUR", "average", month_end) in fx
from brick_consolidation.infrastructure.tables import md
assert not [t for t in md.tables if "rate" in t.replace("_rule", "")] # no rate table here: gl's rates only
# ── CON-33 ─────────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_33_a_structure_row_naming_a_company_gl_does_not_know_is_refused_and_a_report_on_an_unpublished_version_is_refused(gl):
from test_slice_b2 import publish_versions
world(gl)
publish_versions(gl)
body = {"group_id": "GRP", "kind": "acquisition", "company_code": "XX99", "investor_company_code": "UK01", "ownership_pct": "80", "method": "full",
"effective_date": "2026-04-01", "consideration": "1.00"}
status, text = api("POST", "/structure-changes", ACC, body)
assert status == 409 and "which gl does not know" in str(text)
assert ("companies",) in gl.calls
ok(api("POST", "/groups", ACC, {"group_id": "GRP", "fsv_id": "GROUP-IFRS", "version": state()[1]}))
for step_ in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step_}", ACC, {}))
assert api("GET", f"{BASE}/report", ACC, {"fsv_id": "GROUP-IFRS"})[0] == 200
status, text = api("GET", f"{BASE}/report", ACC, {"fsv_id": "UNPUBLISHED"})
assert status == 409 and "not published" in str(text)
assert ("chart", "GROUP") in gl.calls and ("fsv", "GROUP-IFRS") in gl.calls
# ── CON-34 ─────────────────────────────────────────────────────────────────────────────────────────────────────────
RUN = "odoo:ACME:2026-11-01"
DOORS = json.dumps({"group_id": "GRP", "carried_periods": ["2026-10"], "decisions": {"DE01": "full"}, "odoo_chart_id": 7, "top_company_code": "UK01",
"group_chart_id": "GROUP"})
def odoo_world(gl, extra=0, journals_each=1):
odoo = FakeOdoo(extra=extra, journals_each=journals_each)
seams.ports.odoo = odoo
gl_signed(gl, odoo)
return odoo
def rows_of(table):
from sqlalchemy import select
from brick_consolidation.infrastructure import tables as T
return db(lambda r: [dict(x) for x in r.conn.execute(select(getattr(T, table))).mappings().all()])
def test_con_34_the_structure_the_rules_and_the_journals_of_closed_periods_are_written_read_only_keyed_on_the_odoo_record(gl):
odoo_world(gl)
made = ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS}))
assert made["status"] == "loaded"
[grp] = rows_of("cons_group")
assert (grp["group_id"], grp["group_currency"], grp["top_company_code"], grp["group_chart_id"], grp["ledger_id"]) == ("GRP", "GBP", "UK01", "GROUP", "IFRS")
assert {r["group_account_id"]: r["rate_type"] for r in rows_of("cons_account_rule")} == {"G-1100": "closing", "G-3000": "historical", "G-4000": "average"}
structure = {(r["company_code"], r["valid_from_period"]): (format(D(str(r["ownership_pct"])).normalize(), "f"), r["method"]) for r in rows_of("cons_group_structure")}
assert structure == {("DE01", "2026-10"): ("80", "full"), ("DE01", "2026-11"): ("80", "full")} # history, then the door's decision from the cut-over
[gp] = rows_of("cons_group_period")
assert (gp["period_id"], gp["status"], gp["signed_by"], gp["prepared_by"]) == ("2026-10", "signed", "migration", RUN) and len(gp["tb_digest"]) == 64
from brick_consolidation.application.loader import journal_key
heads = {r["cons_journal_id"]: r for r in rows_of("cons_journal")}
assert set(heads) == {journal_key("ACME", 41), journal_key("ACME", 42)} # derived_id("odoo:<db>:consolidation.journal:<id>")
head = heads[journal_key("ACME", 41)]
assert head["document_type"] == "migration" and head["migration_run_id"] == RUN and head["status"] == "posted" and head["entry_no"] == 0
assert sorted(ln["amount"] for ln in rows_of("cons_journal_line")) == [D("-50.00")] * 2 + [D("50.00")] * 2 # Odoo's amounts, to the penny
assert {r["node_id"]: r["amount"] for r in rows_of("cons_statement")} == {"G-1100": D("100.00"), "G-3000": D("-100.00")}
maps = {(r["odoo_model"], r["odoo_id"]): r["natural_key"] for r in rows_of("cons_migration_map")}
assert maps[("consolidation.journal", "41")] == head["cons_journal_id"] and maps[("consolidation.period", "21")] == "2026-10" and len(maps) == 1 + 3 + 1 + 2
# reconcile compares ODOO's own aggregates with what was loaded
ok(step("reconcile-cons-migration", SYSTEM))
checks = {c["check"]: c["status"] for c in json.loads(ok(api("GET", f"{RUNS}/{RUN}", OWNER))["checks_json"]).__iter__().__class__ and json.loads(ok(api("GET", f"{RUNS}/{RUN}", OWNER))["checks_json"])}
assert checks == {"consolidated_balance": "equal", "structure": "equal", "journals": "equal", "proportional": "differs"}
def test_con_34_a_second_run_changes_nothing_and_a_signed_run_is_never_rewritten(gl):
odoo = odoo_world(gl)
ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS}))
tables = ("cons_group", "cons_group_structure", "cons_account_rule", "cons_group_period", "cons_journal", "cons_journal_line", "cons_statement",
"cons_migration_map")
before = {t: rows_of(t) for t in tables}
journal_audit = lambda: [a for a in rows_of("cons_audit") if a["group_id"]] # noqa: E731 (the run's own audit grows with its steps)
audit = journal_audit()
ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})) # the same run again
ok(step("reconcile-cons-migration", SYSTEM))
ok(step("reload-cons-migration", SYSTEM)) # a reload loads the same keys again
assert {t: rows_of(t) for t in tables} == before and journal_audit() == audit
ok(step("reconcile-cons-migration", SYSTEM))
ok(step("explain-cons-difference", ACC, {"reason": "decided at the door"}))
signed = ok(step("sign-cons-reconciliation", OWNER))
odoo.journal_rows.append({"id": 99, "period_id": 21, "company_period_id": 32, "auto_generated": False, "name": "late"})
odoo.line_rows.append({"id": 2000, "journal_id": 99, "account_id": 11, "amount": D("1.00"), "note": ""})
again = ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS}))
assert again["checks_json"] == signed["checks_json"] and {t: rows_of(t) for t in tables} == before # signed: Odoo's late journal is not loaded
# another run on the same periods never touches a period a signed run carried
other = "odoo:ACME:2026-12-01"
ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": other, "doors_json": DOORS}))
assert {t: rows_of(t) for t in tables} == before and len(rows_of("cons_journal")) == 2
def test_con_34_gls_migration_must_be_signed_companies_and_accounts_mapped_by_one_lookup_per_model(gl):
odoo_world(gl, extra=4, journals_each=3)
gl.maps[("ACME", "res.company")].pop("2")
gl.maps[("ACME", "account.account")].pop("103")
status, text = api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})
assert status == 409 and "gl's migration is not signed" in str(text) and "companies 2" in str(text) and "accounts 103" in str(text)
assert rows_of("cons_group") == [] and rows_of("cons_migration_run") == [] and rows_of("cons_cons_migration_runs") == [] # nothing written, no run made
assert [c for c in gl.calls if c[0] == "map"] == [("map", "res.company", 6), ("map", "account.account", 3)] # one IN-list per model, not per id
gl.maps[("ACME", "res.company")]["2"] = "DE01"
gl.maps[("ACME", "account.account")]["103"] = "L103"
ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS}))
assert len(rows_of("cons_journal")) == 18
def test_con_34_the_load_is_in_resumable_chunks_a_stopped_load_resumes_where_it_stopped(gl, monkeypatch):
from brick_consolidation.application import loader
monkeypatch.setattr(loader, "CHUNK", 4)
odoo = odoo_world(gl, extra=4, journals_each=3) # 18 journals: 5 chunks
real = odoo.journal_lines
calls = []
def dies_on_the_third(ids):
calls.append(1)
if len(calls) == 3:
raise RuntimeError("Odoo went away")
return real(ids)
monkeypatch.setattr(odoo, "journal_lines", dies_on_the_third)
try:
api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})
except RuntimeError:
pass
done = len(rows_of("cons_journal"))
assert done == 8 # two chunks committed, the third not
monkeypatch.setattr(odoo, "journal_lines", real)
ok(api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})) # the same call resumes
assert len(rows_of("cons_journal")) == 18 and len({r["cons_journal_id"] for r in rows_of("cons_journal")}) == 18
assert [c for c in odoo.calls if c[0] == "lines"][-2:] == [("lines", 4), ("lines", 2)]
ok(step("reconcile-cons-migration", SYSTEM))
assert all(c["status"] in ("equal", "differs") for c in json.loads(ok(api("GET", f"{RUNS}/{RUN}", OWNER))["checks_json"]))
# ── statement budget: the Odoo load and its gl lookups are flat from 2 to 10 companies ─────────────────────────────────
LOAD_BUDGET = {"load": 60, "load_again": 31} # statements for the whole load across its transactions (measured: the same for 2 and 10 companies; +1 since audit MUST-7: a reload reads the held journals of each chunk to supersede what Odoo corrected)
def load_paths(gl, engine, extra):
from test_statement_budget import measured
odoo_world(gl, extra=extra, journals_each=1)
out = {}
out["load"], _ = measured(engine, "POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "doors_json": DOORS})
out["load_again"], _ = measured(engine, "POST", f"{RUNS}/{RUN}/reload-cons-migration", SYSTEM)
return out
def test_the_odoo_load_is_flat_in_statements_from_2_to_10_companies_and_pinned(gl, engine):
from test_statement_budget import engine_reset
small = load_paths(gl, engine, 0)
engine_reset(engine)
gl2 = type(gl)()
seams.ports.gl = gl2
big = load_paths(gl2, engine, 8)
print("statements", small, big)
assert big == small, (small, big)
assert all(small[k] <= LOAD_BUDGET[k] for k in small), (small, LOAD_BUDGET)