dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_a1.py
"""Slice A1: CON-01 the group structure, CON-02 the acquisition, CON-03 collecting closed company data from gl."""
from decimal import Decimal
import pytest
from conftest import api, tool
from world import ACC, CTRL, CTRL_SAME_EMAIL, CTRL_SAME_SUB, P, acquire, db, group, ok, period_row, sign_period, state, world
from brick_consolidation.domain import consolidation as C
D = Decimal
def test_con_01_the_structure_in_force_decides_scope_and_method(gl):
group(gl)
gl.set_rows("FR01", "2026-03", [("G-1100", "300000.00", "0.00"), ("G-3000", "-300000.00", "0.00")])
acquire("DE01")
acquire("FR01", "30", "150000.00", "equity")
g, _ = state()
now = C.members_at(g, P)
assert (now["DE01"].method, now["DE01"].ownership, now["FR01"].method, now["FR01"].ownership) == ("full", D("80"), "equity", D("30"))
assert now["UK01"].method == "full" and C.effective_ownership("DE01", now) == D("80")
assert set(C.members_at(g, "2026-03")) == {"UK01"} # before the effective period nobody is owned
assert set(C.members_at(g, "2026-04")) == {"UK01", "DE01", "FR01"}
# a company that has no structure row in a later period is not in scope; a method none leaves scope
ok(api("POST", "/structure-changes", ACC, {"group_id": "GRP", "kind": "method_change", "company_code": "FR01", "investor_company_code": "UK01",
"ownership_pct": "30", "method": "none", "effective_date": "2026-09-15"}))
@pytest.mark.parametrize("bad", ["0", "100.000001", "101", "-5", "abc", "10.1234567"])
def test_con_01_ownership_outside_zero_to_100_is_422_never_500(gl, bad):
group(gl)
status, _ = api("POST", "/structure-changes", ACC, {"group_id": "GRP", "kind": "acquisition", "company_code": "DE01", "investor_company_code": "UK01",
"ownership_pct": bad, "method": "full", "effective_date": "2026-04-01", "consideration": "1.00"})
assert status == 422
def test_con_01_company_must_exist_in_gl_and_investor_in_the_group(gl):
group(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)
status, text = api("POST", "/structure-changes", ACC, {**body, "company_code": "DE01", "investor_company_code": "FR01"})
assert status == 409 and "not in the group" in str(text)
assert api("POST", "/structure-changes", ACC, {**body, "company_code": "UK01", "investor_company_code": "DE01"})[0] == 409
def test_con_01_the_proposer_cannot_approve_matched_on_email_and_sub(gl):
group(gl)
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-04-01", "consideration": "900000.00"}))
url = f"/structure-changes/{change['change_id']}/approve-structure-change"
for other in (CTRL_SAME_EMAIL, CTRL_SAME_SUB):
status, text = api("POST", url, other, {})
assert status == 409 and "proposed" in str(text)
assert api("POST", url, ACC, {})[0] == 403 # the accountant does not hold the approval at all
assert state()[0].group_structures == [] # nothing took effect
assert ok(api("POST", url, CTRL, {}))["status"] == "effective"
assert ok(api("POST", url, CTRL, {}))["status"] == "effective" # a repeat changes nothing
assert len(state()[0].group_structures) == 1
def test_con_01_a_change_never_takes_effect_in_a_signed_period(gl):
group(gl)
gl.set_rows("DE01", "2026-06", [("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"), ("G-1100", "1000000.00", "0.00")])
gl.rates[("EUR", "spot", "2026-07-01")] = D("0.860000")
sign_period("2026-06")
body = {"group_id": "GRP", "kind": "acquisition", "company_code": "DE01", "investor_company_code": "UK01", "ownership_pct": "80", "method": "full",
"effective_date": "2026-04-01", "consideration": "900000.00"}
status, text = api("POST", "/structure-changes", ACC, body)
assert status == 409 and "signed" in str(text)
assert api("POST", "/structure-changes", ACC, {**body, "effective_date": "2026-07-01"})[0] == 200
# approval re-checks: a period signed after the proposal
sign_period("2026-07")
pending = db(lambda r: r.structure_changes.ids())[0]
assert api("POST", f"/structure-changes/{pending}/approve-structure-change", CTRL, {})[0] == 409
def test_con_01_method_full_needs_an_acquisition(gl):
group(gl)
status, text = api("POST", "/structure-changes", ACC, {"group_id": "GRP", "kind": "method_change", "company_code": "FR01", "investor_company_code": "UK01",
"ownership_pct": "60", "method": "full", "effective_date": "2026-09-01"})
assert status == 409 and "acquisition" in str(text)
def test_con_02_goodwill_and_nci_to_the_penny_and_the_equity_fixed_at_the_rate_of_the_day(gl):
group(gl)
shown = 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-04-01",
"consideration": "900000.00"}))
assert shown["status"] == "proposed" and shown["change_id"] == "GRP:DE01:2026-04:acquisition"
assert (D(shown["net_assets"]), D(shown["goodwill"]), D(shown["nci"])) == (D("1000000.00"), D("100000.00"), D("200000.00"))
assert (D(shown["goodwill_group"]), D(shown["nci_group"]), D(shown["rate"])) == (D("86000.00"), D("172000.00"), D("0.860000"))
assert db(lambda r: r.data.historical("GRP", ["DE01"])) == [] # nothing fixed until the controller approves
ok(api("POST", f"/structure-changes/{shown['change_id']}/approve-structure-change", CTRL, {}))
hist = {h["group_account_id"]: h for h in db(lambda r: r.data.historical("GRP", ["DE01"]))}
assert (hist["G-3000"]["amount_local"], hist["G-3000"]["amount_group"]) == (D("-500000.00"), D("-430000.00"))
assert (hist["G-3100"]["amount_local"], hist["G-3100"]["amount_group"], hist["G-3100"]["rate"]) == (D("-500000.00"), D("-430000.00"), D("0.860000"))
# the same figures through the agent door
assert tool(CTRL, "get_structure_change", {"change_id": shown["change_id"]})[1]["goodwill"] == "100000.00"
def test_con_02_fair_value_adjustments_with_their_deferred_tax_and_a_bargain_purchase(gl):
group(gl)
shown = 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-04-01", "consideration": "900000.00",
"fair_value_adjustments": [{"description": "land", "amount": "100000.00", "tax_rate_pct": "30"}]}))
assert D(shown["net_assets"]) == D("1070000.00") # 1,000,000 + 100,000 less 30,000 deferred tax
assert (D(shown["goodwill"]), D(shown["nci"])) == (D("44000.00"), D("214000.00"))
gl.set_rows("DE01", "2026-04", [("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"), ("G-1100", "1000000.00", "0.00")])
gl.rates[("EUR", "spot", "2026-05-01")] = D("0.860000")
cheap = 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-05-01", "consideration": "700000.00"}))
assert cheap["bargain_purchase"] == "true" and D(cheap["goodwill"]) == D("-100000.00") # shown for review, not posted
assert 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-06-01",
"consideration": "12345678901234567.00"})[0] == 422
def test_con_03_collect_closed_companies_and_refuse_the_open_one(gl):
world(gl)
gl.set_rows("FR01", "2026-03", [("G-1100", "300000.00", "0.00"), ("G-3000", "-300000.00", "0.00")])
gl.set_rows("FR01", P, [("G-1100", "10.00", "0.00"), ("G-3000", "-10.00", "0.00")], status="open")
acquire("FR01", "30", "150000.00", "equity")
status, text = api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {})
assert status == 409 and "company_periods_closed" in str(text) and "FR01" in str(text)
assert db(lambda r: r.data.reported("GRP", [P], ["UK01", "DE01", "FR01"])) == [] # nothing written
assert period_row().status == "open"
def test_con_03_collect_stores_the_rows_with_gls_digest_and_a_repeat_writes_nothing(gl):
world(gl)
out = ok(api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {}))
assert out["result"]["collected"] == ["DE01", "UK01"] and period_row().status == "collected" and period_row().run_no == 1
rows = db(lambda r: r.data.reported("GRP", [P], ["DE01", "UK01"]))
de = {(r["group_account_id"], r["partner_company"], r["movement_type"]): r for r in rows if r["company_code"] == "DE01"}
assert de[("G-1100", "", "")]["ytd_local"] == D("1400000.00") and de[("G-1100", "", "")]["local_currency"] == "EUR"
assert de[("G-2300", "UK01", "")]["ytd_local"] == D("-997.70") and de[("G-3100", "", "dividend")]["period_local"] == D("40000.00")
assert sum(r["ytd_local"] for r in rows if r["company_code"] == "DE01") == 0
coll = db(lambda r: r.data.collections("GRP", P))
assert {c["company_code"]: c["gl_digest"] for c in coll}["DE01"] == gl.balances("DE01", "IFRS", P)["digest"]
assert all(c["provisional"] == "false" for c in coll)
version = state()[1]
again = ok(api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {}))
assert again["result"] == {"changed": False} and state()[1] == version and db(lambda r: r.data.collections("GRP", P)) == coll
def test_con_03_a_changed_digest_takes_the_next_run_and_marks_later_steps_stale(gl):
world(gl)
base = f"/groups/GRP/group-periods/{P}"
ok(api("POST", f"{base}/collect-group-data", ACC, {}))
ok(api("POST", f"{base}/translate-currencies", ACC, {}))
gl.set_rows("DE01", P, [r if r[0] != "G-1100" else ("G-1100", "1400100.00", "100.00") for r in __import__("world").DE01_ROWS[:1]] + [
("G-2000", "-216916.25", "0.00") if r[0] == "G-2000" else r for r in __import__("world").DE01_ROWS[1:]])
out = ok(api("POST", f"{base}/collect-group-data", ACC, {}))
assert out["result"]["collected"] == ["DE01"] and out["result"]["unchanged"] == ["UK01"]
assert (period_row().status, period_row().run_no, period_row().stale_from_step) == ("collected", 2, "translate")
de = {r["group_account_id"]: r for r in db(lambda r: r.data.reported("GRP", [P], ["DE01"])) if not r["partner_company"] and not r["movement_type"]}
assert de["G-1100"]["ytd_local"] == D("1400100.00")
def test_con_03_unmapped_accounts_and_figures_that_do_not_net_are_refused(gl):
world(gl)
gl.set_rows("DE01", P, [("G-1100", "5.00", "0.00")], unmapped=[{"account_id": 77, "code": "4711", "ytd_amount": "-5.00"}])
status, text = api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {})
assert status == 409 and "4711" in str(text) and "GL-41" in str(text) and "do not net to zero" in str(text)
assert db(lambda r: r.data.collections("GRP", P)) == []
def test_con_03_an_open_period_collects_only_when_the_owner_allows_it_and_is_flagged(gl):
world(gl)
gl.set_rows("DE01", P, __import__("world").DE01_ROWS, status="open")
assert api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {})[0] == 409
with __import__("conftest").use_cases.repository() as repo: # the owner's control; changing it through the door is CON-19
repo.settings.set("allow_open_period_collect", True, "owner")
out = ok(api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {}))
assert out["result"]["provisional"] == ["DE01"]
assert {c["company_code"]: c["provisional"] for c in db(lambda r: r.data.collections("GRP", P))} == {"DE01": "true", "UK01": "false"}
def test_con_03_collect_needs_its_permission_and_gl_is_read_in_the_spec_shape(gl):
world(gl)
from world import AUDITOR
assert api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", AUDITOR, {})[0] == 403
assert api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", CTRL, {})[0] == 403
gl.calls.clear()
ok(api("POST", f"/groups/GRP/group-periods/{P}/collect-group-data", ACC, {}))
assert [c for c in gl.calls if c[0] == "balances"] == [("balances", "DE01"), ("balances", "UK01")] # one read per company