dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_a2.py
"""Slice A2: CON-04 translation, CON-05 intercompany balances, CON-06 revenue/expense and dividend, CON-07 profit in stock."""
from decimal import Decimal
import pytest
from conftest import api, tool
from sqlalchemy.exc import IntegrityError
from world import ACC, CTRL, P, collected, db, ok, period_row, state, translated, world
D = Decimal
BASE = f"/groups/GRP/group-periods/{P}"
def entries(doc=None, status=None):
out = db(lambda r: r.data.entries("GRP", P, 1))
return [e for e in out if (doc is None or e["document_type"] == doc) and (status is None or e["status"] == status)]
def lines(entry):
return {(ln["group_account_id"], ln["company_code"], ln["partner_company"]): ln["amount"] for ln in entry["lines"]}
def trial(run_no=1):
rows = db(lambda r: r.data.translated("GRP", P, run_no))
return {(r["company_code"], r["group_account_id"], r["partner_company"], r["movement_type"]): r for r in rows}
def none_stated(holder="DE01", seller="UK01"):
return ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": holder, "seller_company": seller, "none_stated": True}))
def ic_ready(gl, accepted=True):
none_stated()
gl.recs[("UK01", "DE01")] = {"match_id": "M1", "as_at": "2026-10-31", "matched": [], "run_at": "2026-11-01T00:00:00Z", "breaks": [
{"key": "INV-77", "ours": "870.00", "theirs": "868.00", "currency": "GBP", "difference": "2.00", "within_tolerance": True,
"accepted": accepted, "reason": "bill booked a month later", "fix": "post the bill in gl"}]}
# ── CON-04 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_04_each_account_at_its_rule_and_the_difference_lands_in_the_translation_reserve(gl):
out = translated(gl)
assert period_row().status == "translated" and out["result"]["rows"] == 15
t = trial()
assert t[("DE01", "G-1100", "", "")]["ytd_group"] == D("1218000.00") and t[("DE01", "G-1100", "", "")]["rate"] == D("0.870000") # closing
assert t[("DE01", "G-4000", "", "")]["ytd_group"] == D("-258000.00") and t[("DE01", "G-4000", "", "")]["rate_type"] == "average" # month by month
assert t[("DE01", "G-3000", "", "")]["ytd_group"] == D("-430000.00") and t[("DE01", "G-3000", "", "")]["rate_type"] == "historical"
assert t[("DE01", "G-3100", "", "")]["ytd_group"] + t[("DE01", "G-3100", "", "dividend")]["ytd_group"] == D("-430000.00") # equity: the fixed amount
assert t[("DE01", "G-2300", "UK01", "")]["ytd_group"] == D("-868.00") and t[("DE01", "G-5100", "UK01", "")]["ytd_group"] == D("5000.00")
assert t[("UK01", "G-1300", "DE01", "")]["ytd_group"] == D("870.00") and t[("UK01", "G-1300", "DE01", "")]["rate"] == D("1.000000")
[entry] = entries("translation", "posted")
de = sum(r["ytd_group"] for k, r in t.items() if k[0] == "DE01")
assert de == D("11941.86")
assert lines(entry) == {("G-3900", "DE01", ""): D("-9553.49"), ("G-3950", "DE01", ""): D("-2388.37")} # 80% / 20% of the difference
assert de + sum(entry["lines"][i]["amount"] for i in range(len(entry["lines"]))) == 0 # exactly, no float tail
assert sum(r["ytd_group"] for k, r in t.items() if k[0] == "UK01") == 0
def test_con_04_a_missing_rate_refuses_naming_currency_and_month_and_writes_nothing(gl):
collected(gl)
del gl.rates[("EUR", "average", "2026-07")]
status, text = api("POST", f"{BASE}/translate-currencies", ACC, {})
assert status == 409 and "EUR" in str(text) and "2026-07" in str(text) and "GL-33" in str(text)
assert period_row().status == "collected" and db(lambda r: r.data.translated("GRP", P, 1)) == [] and entries() == []
gl.rates[("EUR", "average", "2026-07")] = D("0.860000")
del gl.rates[("EUR", "closing", "2026-10-31")]
assert "closing rate EUR" in str(api("POST", f"{BASE}/translate-currencies", ACC, {})[1])
def test_con_04_translating_twice_writes_nothing_and_a_new_collection_replaces_the_run_without_doubling(gl):
translated(gl)
version = state()[1]
assert ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))["result"] == {"changed": False} and state()[1] == version
assert len(entries("translation")) == 1
# gl posts more for DE01: the digest changes, the next runs take run 2
gl.set_rows("DE01", P, [("G-1100", "1400100.00", "100.00") if r[0] == "G-1100" else ("G-2000", "-216916.25", "0.00") if r[0] == "G-2000" else r
for r in __import__("world").DE01_ROWS])
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
assert period_row().run_no == 2 and period_row().stale_from_step == ""
out = {e["run_no"]: e for e in db(lambda r: r.data.entries("GRP", P, 2)) if e["document_type"] == "translation" and not e["reverses_id"]}
assert out[1]["status"] == "reversed" and out[2]["status"] == "posted"
[mirror] = [e for e in db(lambda r: r.data.entries("GRP", P, 2)) if e["reverses_id"]]
assert mirror["reverses_id"] == out[1]["cons_journal_id"] and {k: -v for k, v in lines(out[1]).items()} == lines(mirror)
assert len(db(lambda r: r.data.translated("GRP", P, 1))) == len(db(lambda r: r.data.translated("GRP", P, 2))) == 15 # run 1 kept
# the books of the period (posted + reversed originals + mirrors) hold run 2 only
net = {}
for e in db(lambda r: r.data.entries("GRP", P, 2)):
for ln in e["lines"]:
net[ln["group_account_id"]] = net.get(ln["group_account_id"], 0) + ln["amount"]
assert net == {k[0]: v for k, v in lines(out[2]).items()}
def test_con_04_two_runs_of_one_period_collide_on_one_key(gl):
collected(gl)
from brick_consolidation.application import use_cases
with use_cases.repository() as repo:
repo.data.claim("GRP", P, "translate", 1) # another translate of run 1 got there first
assert api("POST", f"{BASE}/translate-currencies", ACC, {})[0] == 409
with pytest.raises(IntegrityError), use_cases.repository() as repo:
repo.data.claim("GRP", P, "translate", 1)
assert period_row().status == "collected"
def test_con_04_translate_needs_collect_first_and_its_own_permission(gl):
world(gl)
assert api("POST", f"{BASE}/translate-currencies", ACC, {})[0] == 409
collected_ok = ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
assert collected_ok["result"]["changed"]
assert api("POST", f"{BASE}/translate-currencies", CTRL, {})[0] == 403
assert tool(ACC, "translate_currencies", {"group_id": "GRP", "period_id": P})[0] == "ok"
# ── CON-05 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_05_receivable_against_payable_removes_both_sides_and_the_accepted_break_goes_to_ic_difference(gl):
translated(gl)
ic_ready(gl)
out = ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
assert period_row().status == "eliminated" and len(out["result"]["entries"]) >= 1
[entry] = entries("ic_balance", "posted")
assert lines(entry) == {("G-1300", "UK01", "DE01"): D("-870.00"), ("G-2300", "DE01", "UK01"): D("868.00"), ("G-1990", "UK01", "DE01"): D("2.00")}
assert sum(ln["amount"] for ln in entry["lines"]) == 0
assert ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))["result"] == {"changed": False} # a repeat writes nothing
assert len(entries("ic_balance")) == 1
def test_con_05_above_tolerance_the_pair_is_refused_with_gls_breaks_and_fix_then_run_2_replaces(gl):
translated(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 __import__("world").UK01_ROWS])
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
ic_ready(gl, accepted=False)
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 "post the bill in gl" in str(text)
assert period_row().status == "translated" and entries("ic_balance") == []
# fixed in gl, collected again (run 3), translated, eliminated: one set of entries, the refused run left none
gl.set_rows("UK01", P, __import__("world").UK01_ROWS)
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
gl.recs[("UK01", "DE01")]["breaks"][0].update({"difference": "2.00", "within_tolerance": True, "accepted": True})
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
assert period_row().run_no == 3 and [e["run_no"] for e in db(lambda r: r.data.entries("GRP", P, 3)) if e["document_type"] == "ic_balance"] == [3]
def test_con_05_a_difference_gl_does_not_explain_is_translation_and_goes_to_the_cta(gl):
translated(gl)
none_stated() # gl ran no match: no breaks to explain the 2.00
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[entry] = entries("ic_balance")
assert lines(entry)[("G-3900", "UK01", "DE01")] == D("2.00") and ("G-1990", "UK01", "DE01") not in lines(entry)
# ── CON-06 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_06_revenue_against_cost_and_the_dividend_with_the_nci_bearing_its_share(gl):
translated(gl)
ic_ready(gl)
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[pl] = entries("ic_pl")
assert lines(pl) == {("G-4100", "UK01", "DE01"): D("5000.00"), ("G-5100", "DE01", "UK01"): D("-5000.00")} and sum(ln["amount"] for ln in pl["lines"]) == 0
[div] = entries("dividend")
assert lines(div) == {("G-4300", "UK01", "DE01"): D("27520.00"), ("G-3100", "DE01", "UK01"): D("-34400.00"), ("G-3950", "DE01", ""): D("6880.00")}
assert sum(ln["amount"] for ln in div["lines"]) == 0 # 80% of the dividend is UK01's income; 20% the NCI's
def test_con_06_the_difference_in_the_p_and_l_goes_to_ic_difference_pl(gl):
translated(gl)
ic_ready(gl)
gl.set_rows("DE01", P, [("G-5100", "5800.00", "800.00", "UK01") if r[0] == "G-5100" else ("G-2000", "-216802.30", "0.00") if r[0] == "G-2000" else r
for r in __import__("world").DE01_ROWS])
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[pl] = [e for e in db(lambda r: r.data.entries("GRP", P, 2)) if e["document_type"] == "ic_pl" and e["status"] == "posted" and not e["reverses_id"]]
assert ("G-6900", "UK01", "DE01") in lines(pl) and sum(ln["amount"] for ln in pl["lines"]) == 0
# ── CON-07 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_07_unrealised_profit_in_stock_with_deferred_tax_at_the_holders_rate(gl):
translated(gl)
ic_ready(gl)
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "inventory_amount": "12000.00", "margin_pct": "25",
"tax_rate_pct": "30"}))
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[entry] = entries("stock_profit")
assert lines(entry) == {("G-5000", "UK01", "DE01"): D("2610.00"), ("G-1400", "DE01", "UK01"): D("-2610.00"),
("G-1450", "DE01", "UK01"): D("783.00"), ("G-6500", "DE01", "UK01"): D("-783.00")} # 3,000.00 EUR and 900.00 EUR at 0.87
def test_con_07_an_upstream_sale_from_a_partly_owned_seller_lets_the_nci_bear_its_share(gl):
translated(gl)
ic_ready(gl)
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "UK01", "seller_company": "DE01", "inventory_amount": "10000.00", "margin_pct": "20",
"tax_rate_pct": "25"}))
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
[entry] = entries("stock_profit")
got = lines(entry)
assert got[("G-5000", "DE01", "UK01")] == D("2000.00") and got[("G-1450", "UK01", "DE01")] == D("500.00")
assert got[("G-3950", "DE01", "")] == D("300.00") and got[("G-3960", "DE01", "")] == D("-300.00") # 20% of (2,000 - 500)
assert sum(ln["amount"] for ln in entry["lines"]) == 0
def test_con_07_a_selling_pair_without_an_inventory_entry_is_listed_and_bounds_are_422(gl):
translated(gl)
status, text = api("POST", f"{BASE}/run-eliminations", ACC, {})
assert status == 409 and "ic_inventory_entered" in str(text) and "UK01" in str(text) and "DE01" in str(text)
row = {"holder_company": "DE01", "seller_company": "UK01", "inventory_amount": "12000.00", "margin_pct": "25", "tax_rate_pct": "30"}
for bad in ({"margin_pct": "100.5"}, {"margin_pct": "-1"}, {"inventory_amount": "12345678901234567.00"}, {"inventory_amount": "1.234"},
{"tax_rate_pct": "101"}):
assert api("POST", f"{BASE}/ic-inventory", ACC, {**row, **bad})[0] == 422
ok(api("POST", f"{BASE}/ic-inventory", ACC, row))
assert ok(api("POST", f"{BASE}/ic-inventory", ACC, row))["version"] > 0
assert api("POST", f"{BASE}/ic-inventory", CTRL, row)[0] == 403
assert len(db(lambda r: r.data.inventory("GRP", P))) == 1
def test_eliminations_need_an_approved_acquisition_for_every_full_company(gl):
world(gl, acquired=False, opened=False)
from world import group_structures_without_acquisition
group_structures_without_acquisition()
ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
status, text = api("POST", f"{BASE}/run-eliminations", ACC, {})
assert status == 409 and "no approved acquisition for DE01" in str(text)