dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_b1.py
"""Slice B1: CON-08 investments and the minority, CON-09 the equity method, CON-10 manual entries with numbers."""
from decimal import Decimal
import pytest
from conftest import ALL, api, as_role, person, tool
from sqlalchemy import insert
from sqlalchemy.exc import IntegrityError
from world import ACC, CTRL, CTRL_SAME_EMAIL, CTRL_SAME_SUB, P, acquire, db, gl_world, group, ok, state, world
from brick_consolidation.domain import consolidation as C
D = Decimal
BASE = f"/groups/GRP/group-periods/{P}"
def entries(doc=None, status=None, run=1):
out = db(lambda r: r.data.entries("GRP", P, run))
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):
"""The entry's lines summed per (account, company, partner)."""
out = {}
for ln in entry["lines"]:
k = (ln["group_account_id"], ln["company_code"], ln["partner_company"])
out[k] = out.get(k, D(0)) + ln["amount"]
return out
def tb():
from brick_consolidation.application import reports
g, _ = state()
gp = next(x for x in g.group_periods if x.period_id == P)
return db(lambda r: reports.consolidated(r, g, gp)[0])
def add_rules(*rows, roles=()):
ok(api("POST", "/groups/GRP/rules", ACC, {"account_rules": [{"group_account_id": a, "valid_from_period": "2020-01", "rate_type": r} for a, r in rows],
"account_roles": [{"role": r, "valid_from_period": "2020-01", "group_account_id": a} for r, a in roles]}))
# ── CON-08 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def exact_world(gl):
"""UK01 bought 80% of DE01 for 900,000.00 EUR; DE01's 2026 result to October is 60,000.00 EUR; nothing intercompany."""
world(gl)
gl.set_rows("DE01", P, [("G-1100", "1060000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-60000.00", "-10000.00")])
gl.set_rows("UK01", P, [("G-1600", "774000.00", "0.00", "DE01"), ("G-3000", "-774000.00", "0.00")])
for step in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step}", ACC, {}))
def eliminate():
return ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
def test_con_08_investment_eliminated_against_80_percent_of_the_equity_with_goodwill_and_the_nci(gl):
exact_world(gl)
out = eliminate()["result"]
[inv] = entries("investment")
assert lines(inv) == {("G-1600", "UK01", "DE01"): D("-774000.00"), ("G-3000", "DE01", ""): D("344000.00"),
("G-3100", "DE01", ""): D("344000.00"), ("G-1700", "DE01", ""): D("87000.00"),
("G-3900", "DE01", ""): D("-1000.00")} # 80% of 860,000.00; goodwill 100,000.00 EUR at the closing 0.87; the 1,000.00 gain to the reserve (IAS 21.47)
assert sum(ln["amount"] for ln in inv["lines"]) == 0
[nci] = entries("nci")
assert lines(nci) == {("G-3000", "DE01", ""): D("86000.00"), ("G-3100", "DE01", ""): D("86000.00"), ("G-3950", "DE01", ""): D("-182320.00"),
("G-3960", "DE01", ""): D("10320.00")} # 20% of 860,000 at acquisition, and 20% of the 51,600.00 GBP result
assert sum(ln["amount"] for ln in nci["lines"]) == 0
assert D(out["investments"][0]["eliminated"]) == D("774000.00") and D(out["investments"][0]["post_acquisition_result"]) == D("51600.00")
books = tb()
assert books.get(("UK01", "G-1600", "DE01", "", ""), D(0)) == 0 # the gate investment_eliminated: exactly zero
assert not [k for k in books if k[0] == "DE01" and k[1] in ("G-3000", "G-3100")] # the subsidiary's equity is gone
assert books[("DE01", "G-3950", "", "", "")] == D("-184440.00") and books[("DE01", "G-1700", "", "", "")] == D("87000.00")
assert sum(books.values()) == 0 # the group balances, exactly
nci_total = -books[("DE01", "G-3950", "", "", "")] - books[("DE01", "G-3960", "", "", "")]
assert nci_total == D("0.20") * (D("860000.00") + D("51600.00") + D("10600.00")) - D("10320.00") # 20% of net assets: equity, result, translation
def test_con_08_goodwill_is_not_amortised_under_ifrs_and_is_over_the_life_where_the_group_sets_one(gl):
exact_world(gl)
eliminate()
assert not [k for k in tb() if k[1] == "G-6600"] # no life set: no amortisation
ok(api("POST", "/groups", ACC, {"group_id": "GRP", "goodwill_life_years": "10", "version": state()[1]}))
add_rules(("G-6600", "average"), roles=[("goodwill_amortisation", "G-6600"), ("retained_earnings", "G-3100")])
# recollect is not needed: only the group's setting changed, so eliminate anew through a new run
gl.set_rows("DE01", P, [("G-1100", "1060000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-60000.00", "-10000.00")][:3] + [("G-4000", "-60000.00", "-9000.00"), ])
# DE01 no longer nets to zero: the collection is refused; put the figures back with a different period split instead
gl.set_rows("DE01", P, [("G-1100", "1060000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-60000.00", "-9000.00")])
ok(api("POST", f"{BASE}/recollect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
out = eliminate()["result"]
assert D(out["investments"][0]["goodwill_amortised"]) == D("5075.00") # 87,000.00 (at the closing rate) x 7 / 120, April to October
[inv] = [e for e in entries("investment", "posted", 2) if not e["reverses_id"]]
assert lines(inv)[("G-1700", "DE01", "")] == D("81925.00") and lines(inv)[("G-6600", "DE01", "")] == D("5075.00")
assert sum(ln["amount"] for ln in inv["lines"]) == 0
def test_con_08_rerun_replaces_and_the_investment_stays_at_zero(gl):
exact_world(gl)
eliminate()
gl.set_rows("DE01", P, [("G-1100", "1070000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-70000.00", "-20000.00")])
ok(api("POST", f"{BASE}/recollect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
eliminate()
posted = [e for e in db(lambda r: r.data.entries("GRP", P, 2)) if e["document_type"] in ("investment", "nci") and not e["reverses_id"]]
assert sorted((e["document_type"], e["run_no"], e["status"]) for e in posted) == [("investment", 1, "reversed"), ("investment", 2, "posted"),
("nci", 1, "reversed"), ("nci", 2, "posted")]
assert tb().get(("UK01", "G-1600", "DE01", "", ""), D(0)) == 0 and sum(tb().values()) == 0
def test_con_08_refusals_name_the_gate(gl):
exact_world(gl)
gl.set_rows("UK01", P, [("G-3000", "0.00", "0.00")])
ok(api("POST", f"{BASE}/recollect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
status, text = api("POST", f"{BASE}/run-eliminations", ACC, {})
assert status == 409 and "investment_eliminated" in str(text) and "G-1600" in str(text) and "DE01" in str(text)
assert entries("investment") == [] # nothing is half-written
def test_con_08_a_purchase_part_way_through_the_year_takes_out_the_months_before_it(gl):
"""Audit v1 MUST-1a: this was refused (post_acquisition_result); the months before the purchase are now taken out of the result, each at its
own month's average rate, from gl's movement of the month. Nothing was earned before July here, so the whole result is post-acquisition."""
gl_world(gl)
group(gl)
gl.set_rows("DE01", "2026-06", [("G-1100", "1000000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00")])
gl.rates[("EUR", "spot", "2026-07-01")] = D("0.860000")
acquire(date_="2026-07-01")
ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
gl.set_rows("UK01", P, [("G-1600", "774000.00", "0.00", "DE01"), ("G-3000", "-774000.00", "0.00")])
gl.set_rows("DE01", P, [("G-1100", "1060000.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-60000.00", "-10000.00")])
for step in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step}", ACC, {}))
[inv] = eliminate()["result"]["investments"]
pl = [r["ytd_group"] for r in db(lambda r: r.data.translated("GRP", P, state()[0].group_periods[-1].run_no))
if r["company_code"] == "DE01" and r["group_account_id"] == "G-4000"]
assert D(inv["post_acquisition_result"]) == -sum(pl) != 0 and D(inv["nci_result"]) == C.q2(-sum(pl) * D("0.2"))
# ── CON-09 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def associate_world(gl):
world(gl, acquired=False, opened=False)
gl.set_rows("FR01", "2026-03", [("G-1100", "300000.00", "0.00"), ("G-3000", "-300000.00", "0.00")])
acquire("FR01", "30", "300000.00", method="equity")
ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
gl.set_rows("FR01", P, [("G-1100", "344000.00", "0.00"), ("G-3000", "-300000.00", "0.00"), ("G-3100", "6000.00", "6000.00", "", "dividend"),
("G-4000", "-50000.00", "-8000.00")])
gl.set_rows("UK01", P, [("G-1600", "258000.00", "0.00", "FR01"), ("G-4300", "-5160.00", "-5160.00", "FR01"), ("G-3000", "-252840.00", "0.00")])
add_rules(("G-4500", "average"), ("G-3970", "historical"), roles=[("share_of_associates", "G-4500"), ("oci", "G-3970")])
for step in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step}", ACC, {}))
def test_con_09_the_associate_is_one_entry_not_line_by_line(gl):
associate_world(gl)
out = eliminate()["result"]
assert entries("translation") == [] # an associate has no translation entry of its own
[em] = entries("equity_method")
assert lines(em) == {("G-1600", "UK01", "FR01"): D("12900.00") + D("1050.00") - D("5160.00"), ("G-4500", "UK01", "FR01"): D("-12900.00"),
("G-3970", "UK01", "FR01"): D("-1050.00"), ("G-4300", "UK01", "FR01"): D("5160.00")}
assert sum(ln["amount"] for ln in em["lines"]) == 0
[shown] = out["associates"]
assert D(shown["share_of_result_local"]) == D("15000.00") and D(shown["share_of_result"]) == D("12900.00") # 30% of 50,000.00 EUR, then at 0.86
assert D(shown["oci"]) == D("-1050.00") and D(shown["dividend"]) == D("5160.00")
books = tb()
assert not [k for k in books if k[0] == "FR01"] and sum(books.values()) == 0
assert books[("UK01", "G-1600", "FR01", "", "")] == D("266790.00") and ("UK01", "G-4300", "FR01", "", "") not in books # carried at the share
assert eliminate_again() == {"changed": False}
def eliminate_again():
return ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))["result"]
def test_con_09_rerun_after_a_new_collection_replaces(gl):
associate_world(gl)
eliminate()
gl.set_rows("FR01", P, [("G-1100", "354000.00", "0.00"), ("G-3000", "-300000.00", "0.00"), ("G-3100", "6000.00", "6000.00", "", "dividend"),
("G-4000", "-60000.00", "-18000.00")])
ok(api("POST", f"{BASE}/recollect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))
got = [(e["run_no"], e["status"]) for e in db(lambda r: r.data.entries("GRP", P, 2)) if e["document_type"] == "equity_method" and not e["reverses_id"]]
assert sorted(got) == [(1, "reversed"), (2, "posted")]
assert tb()[("UK01", "G-4500", "FR01", "", "")] == D("-15480.00") # 30% of 60,000.00 EUR at 0.86
# ── CON-10 ─────────────────────────────────────────────────────────────────────────────────────────────────────
IMPAIRMENT = [{"group_account_id": "G-6600", "company_code": "DE01", "amount": "10000.00"},
{"group_account_id": "G-1700", "company_code": "DE01", "amount": "-10000.00"}]
def propose(key="E1", lines_=None, headers=ACC, **more):
return api("POST", "/cons-journals/propose-cons-entry", headers, {"cons_journal_id": key, "group_id": "GRP", "period_id": P, "memo": "impairment",
"lines": lines_ or IMPAIRMENT, **more})
def manual_world(gl):
world(gl)
add_rules(("G-6600", "average"))
ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))
def numbers():
from brick_consolidation.infrastructure.tables import cons_number_series
return db(lambda r: sorted((x["fiscal_year"], x["entry_no"]) for x in r.conn.execute(cons_number_series.select()).mappings().all()))
def test_con_10_a_goodwill_impairment_is_proposed_approved_and_numbered_in_the_books_year(gl):
manual_world(gl)
made = ok(propose())
jid = made["cons_journal_id"]
assert made["status"] == "proposed" and made["document_type"] == "manual" and made["proposed_by"] == "group_accountant@group.example"
assert ok(propose())["cons_journal_id"] == jid and ok(propose())["version"] == made["version"] # the same key is the same entry
assert numbers() == [] # proposing takes no number
posted = ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))
assert posted["status"] == "posted" and posted["entry_no"] == 1 and posted["approved_by"] == "group_controller@group.example"
assert numbers() == [("FY2027", 1)] # the year the group's financial year ends in
assert ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))["version"] == posted["version"] # a repeat changes nothing
assert tool(CTRL, "get_cons_journal", {"cons_journal_id": jid})[0] == "ok"
def test_con_10_the_proposer_cannot_approve_under_any_identity(gl):
manual_world(gl)
jid = ok(propose())["cons_journal_id"]
same = person("group_accountant@group.example", ALL)
assert api("POST", f"/cons-journals/{jid}/approve-cons-entry", same, {})[0] == 409 # the same email and sub
for other in (CTRL_SAME_EMAIL, CTRL_SAME_SUB): # one identity matches: still the proposer
status, text = api("POST", f"/cons-journals/{jid}/approve-cons-entry", other, {})
assert status == 409 and "proposed the entry" in str(text), (status, text)
assert api("POST", f"/cons-journals/{jid}/approve-cons-entry", ACC, {})[0] == 403 # an accountant holds no approval
assert numbers() == []
ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))
assert api("POST", "/cons-journals/propose-cons-entry", CTRL, {"cons_journal_id": "X", "group_id": "GRP", "period_id": P, "lines": IMPAIRMENT})[0] == 403
def person_with_everything():
from conftest import ALICE
return ALICE # alice holds both permissions; she proposes first in this test
def test_con_10_the_second_person_may_be_switched_off_only_by_the_owner(gl):
manual_world(gl)
status, body = api("POST", "/settings/signoff_second_person", as_role("group_accountant"), {"value": False, "valid_from": "2026-11-20"})
assert status == 403 and "control" in str(body) # the accountant (cons:settings:write) cannot switch it off for her own entries
ok(api("POST", "/controls/signoff_second_person", as_role("owner"), {"value": False, "valid_from": "2026-11-20"})) # the first row of a key may start today
jid = ok(propose(headers=person_with_everything()))["cons_journal_id"]
assert ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", person_with_everything(), {}))["status"] == "posted"
def test_con_10_validation_balanced_scope_partner_movement_and_bounds(gl):
manual_world(gl)
unbalanced = [IMPAIRMENT[0], {**IMPAIRMENT[1], "amount": "-9999.99"}]
status, text = propose("U", unbalanced)
assert status == 409 and "not balanced" in str(text)
assert propose("S", [{**IMPAIRMENT[0], "company_code": "XX99"}, IMPAIRMENT[1]])[0] == 409 # not in scope
assert propose("I", [{"group_account_id": "G-1300", "company_code": "UK01", "amount": "5.00"}, {"group_account_id": "G-2300", "company_code": "DE01",
"partner_company": "UK01", "amount": "-5.00"}])[0] == 409 # partner
assert propose("M", [{"group_account_id": "G-3000", "company_code": "DE01", "amount": "5.00"}, {"group_account_id": "G-1700", "company_code": "DE01",
"amount": "-5.00"}])[0] == 409 # movement
ok(propose("M2", [{"group_account_id": "G-3000", "company_code": "DE01", "amount": "5.00", "movement_type": "capital_change"}, IMPAIRMENT[1] | {"amount": "-5.00"}]))
for bad in ("1.234", "12345678901234567.00", "abc"):
assert propose("B", [{**IMPAIRMENT[0], "amount": bad}, IMPAIRMENT[1]])[0] == 422, bad
assert propose("Z", [IMPAIRMENT[0]])[0] == 422 # fewer than two lines
assert propose("G", IMPAIRMENT, group_id="NOPE")[0] == 404
def test_con_10_numbers_are_above_the_high_water_mark_and_a_rejected_entry_takes_none(gl):
manual_world(gl)
a, b, c = (ok(propose(k))["cons_journal_id"] for k in ("A", "B", "C"))
ok(api("POST", f"/cons-journals/{b}/reject-cons-entry", CTRL, {"reason": "not supported"}))
assert ok(api("POST", f"/cons-journals/{a}/approve-cons-entry", CTRL, {}))["entry_no"] == 1
assert ok(api("POST", f"/cons-journals/{c}/approve-cons-entry", CTRL, {}))["entry_no"] == 2 # no number for the rejected one
assert api("POST", f"/cons-journals/{b}/approve-cons-entry", CTRL, {})[0] == 409
from brick_consolidation.infrastructure.tables import cons_number_series
with db_conn() as conn: # a higher number was claimed once
conn.execute(insert(cons_number_series).values(group_id="GRP", fiscal_year="FY2027", entry_no=9, cons_journal_id="gone"))
d = ok(propose("D"))["cons_journal_id"]
assert ok(api("POST", f"/cons-journals/{d}/approve-cons-entry", CTRL, {}))["entry_no"] == 10 # never filling 3..8
with pytest.raises(IntegrityError), db_conn() as conn: # a racer claiming number 10
conn.execute(insert(cons_number_series).values(group_id="GRP", fiscal_year="FY2027", entry_no=10, cons_journal_id="racer"))
from erp_platform.repository import CLAIM_TABLES
assert "cons_number_series" in CLAIM_TABLES # a duplicate at commit is a 409
def db_conn():
from contextlib import contextmanager
from brick_consolidation.application import use_cases
@contextmanager
def conn():
with use_cases.repository() as repo:
yield repo.conn
return conn()
def test_con_10_period_specific_reverses_at_the_next_period_and_cumulative_carries(gl):
manual_world(gl)
one = ok(propose("PS", period_specific=True))["cons_journal_id"]
two = ok(propose("CU", period_specific=False))["cons_journal_id"]
for j in (one, two):
ok(api("POST", f"/cons-journals/{j}/approve-cons-entry", CTRL, {}))
months = C.fiscal_months("2026-11", "3")[:-1]
here = db(lambda r: r.data.books_entries("GRP", P, []))
assert {e["cons_journal_id"] for e in here} == {one, two}
nxt = db(lambda r: r.data.books_entries("GRP", "2026-11", months))
assert [e["cons_journal_id"] for e in nxt] == [two] # the period-specific one is not carried: it reversed
other_year = db(lambda r: r.data.books_entries("GRP", "2027-04", C.fiscal_months("2027-04", "3")[:-1]))
assert other_year == [] # a new financial year starts clean (its carry-forward is CON-15)
def test_con_10_a_correction_is_a_reversal_never_an_edit(gl):
manual_world(gl)
jid = ok(propose())["cons_journal_id"]
assert api("POST", f"/cons-journals/{jid}/reverse-cons-entry", ACC, {"reason": "x"})[0] == 409 # a proposed entry is not posted yet
ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))
assert api("POST", f"/cons-journals/{jid}/reverse-cons-entry", ACC, {"reason": ""})[0] in (409, 422)
done = ok(api("POST", f"/cons-journals/{jid}/reverse-cons-entry", ACC, {"reason": "wrong company"}))
assert done["status"] == "reversed"
[orig, mirror] = sorted(entries("manual"), key=lambda e: e["entry_no"])
assert mirror["reverses_id"] == jid and mirror["entry_no"] == 2 and {k: -v for k, v in lines(orig).items()} == lines(mirror)
assert ok(api("POST", f"/cons-journals/{jid}/reverse-cons-entry", ACC, {"reason": "again"}))["version"] == done["version"]
assert len(entries("manual")) == 2 and sum(v for e in entries("manual") for v in lines(e).values()) == 0 # neither deleted; they net to zero
assert api("POST", f"/cons-journals/{mirror['cons_journal_id']}/reverse-cons-entry", ACC, {"reason": "undo"})[0] == 409
def test_con_10_a_signed_period_refuses_propose_approve_and_reverse(gl):
manual_world(gl)
waiting = ok(propose("W"))["cons_journal_id"]
posted = ok(propose("PP"))["cons_journal_id"]
ok(api("POST", f"/cons-journals/{posted}/approve-cons-entry", CTRL, {}))
_sign_this_period()
assert propose("N")[0] == 409
assert api("POST", f"/cons-journals/{waiting}/approve-cons-entry", CTRL, {})[0] == 409
assert api("POST", f"/cons-journals/{posted}/reverse-cons-entry", ACC, {"reason": "late"})[0] == 409
def _sign_this_period():
from brick_consolidation.application import use_cases
with use_cases.repository() as repo:
g, v = repo.groups.get("GRP")
next(x for x in g.group_periods if x.period_id == P).status = "signed"
repo.groups.save(g, v)
def test_one_claim_key_for_a_manual_entry_and_a_run_of_the_same_group_period(gl):
manual_world(gl)
g, v = state()
jid = ok(propose())["cons_journal_id"] # proposing claims the group's next version
assert state()[1] == v + 1
ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))
assert state()[1] == v + 2
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
assert state()[1] == v + 3
from brick_consolidation.application import use_cases
with pytest.raises(Exception) as late, use_cases.repository() as repo: # a run that loaded the group before the entry was approved
repo.groups.save(g, v + 1)
assert "changed since" in str(late.value) or "Conflict" in type(late.value).__name__