dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_c2.py
"""Slice C2: CON-18 migration-run sign-off, CON-19 settings vs controls, CON-20 re-collecting after a gl period reopens, CON-21 statements."""
from datetime import datetime, timezone
from decimal import Decimal
from conftest import api, as_role
from erp_platform import work
from test_slice_b1 import BASE, eliminate, exact_world
from test_slice_b2 import CTRL2, ready_world
from world import ACC, AUDITOR, CTRL, OWNER, P, db, ok, state, world
D = Decimal
SYSTEM = as_role("system")
# ═══ CON-19 settings vs controls ═══════════════════════════════════════════════════════════════════════════════════════
def test_con_19_the_accountant_is_refused_a_control_and_the_owners_row_starts_tomorrow(gl):
status, body = api("POST", "/settings/signoff_second_person", ACC, {"value": False})
assert status == 403 and body["gate"] == "settings_control" and "owner" in body["fix"]
assert api("POST", "/controls/signoff_second_person", ACC, {"value": False})[0] == 403 # and she does not hold the owner's permission
assert api("POST", "/controls/signoff_second_person", CTRL, {"value": False})[0] == 403
assert api("POST", "/controls/translation_pl_rate", OWNER, {"value": "ytd_average"})[0] == 409 # not a control: the accountant's setting
# what the accountant may set: a day-to-day setting, from tomorrow
assert ok(api("POST", "/settings/translation_pl_rate", ACC, {"value": "ytd_average"}))["valid_from"] == "2026-11-21"
row = ok(api("POST", "/controls/allow_open_period_collect", OWNER, {"value": True}))
assert row["valid_from"] == "2026-11-21" and row["set_by"] == "owner@group.example" # tomorrow: the business date is 2026-11-20
now = {s["setting_key"]: s["value"] for s in ok(api("GET", "/settings", ACC))["settings"]}
assert now["allow_open_period_collect"] is None and now["translation_pl_rate"] is None # not in force today
# never today or earlier once the key has a row; the first row of a key may start today
assert api("POST", "/controls/allow_open_period_collect", OWNER, {"value": False, "valid_from": "2026-11-20"})[0] == 409
assert api("POST", "/controls/allow_open_period_collect", OWNER, {"value": False, "valid_from": "2026-11-01"})[0] == 409
assert ok(api("POST", "/controls/signoff_second_person", OWNER, {"value": True, "valid_from": "2026-11-20"}))["valid_from"] == "2026-11-20"
assert api("POST", "/controls/allow_open_period_collect", OWNER, {"value": "yes"})[0] == 422
assert api("POST", "/controls/allow_open_period_collect", OWNER, {"value": True, "valid_from": "2026-11-21"})[0] == 409 # the same day twice
work.set_clock(at=datetime(2026, 11, 21, 9, 0, tzinfo=timezone.utc), freeze=True)
now = {s["setting_key"]: s["value"] for s in ok(api("GET", "/settings", ACC))["settings"]}
assert now["allow_open_period_collect"] is True and now["translation_pl_rate"] == "ytd_average"
ok(api("POST", "/controls/allow_open_period_collect", OWNER, {"value": False})) # a later change is another row
history = ok(api("GET", "/settings/allow_open_period_collect/history", AUDITOR))["rows"]
assert [(r["valid_from"], r["value"], r["set_by"]) for r in history] == [("2026-11-21", True, "owner@group.example"),
("2026-11-22", False, "owner@group.example")]
def test_con_19_today_is_the_business_date_not_the_utc_date(gl, monkeypatch):
monkeypatch.setenv("BUSINESS_TIMEZONE", "Pacific/Auckland")
work.set_clock(at=datetime(2026, 11, 20, 23, 30, tzinfo=timezone.utc), freeze=True) # already the 21st in Auckland
assert ok(api("POST", "/controls/allow_open_period_collect", OWNER, {"value": True}))["valid_from"] == "2026-11-22"
assert ok(api("GET", "/settings", ACC))["today"] == "2026-11-21"
def test_con_19_the_control_takes_effect_on_the_group_close(gl):
world(gl)
from world import DE01_ROWS
gl.set_rows("DE01", P, DE01_ROWS, status="open")
path = f"/groups/GRP/group-periods/{P}/collect-group-data"
assert api("POST", path, ACC, {})[0] == 409
ok(api("POST", "/controls/allow_open_period_collect", OWNER, {"value": True}))
assert api("POST", path, ACC, {})[0] == 409 # tomorrow, not today
work.set_clock(at=datetime(2026, 11, 21, 9, 0, tzinfo=timezone.utc), freeze=True)
assert ok(api("POST", path, ACC, {}))["result"]["provisional"] == ["DE01"]
# ═══ CON-20 re-collecting after a gl period reopens ══════════════════════════════════════════════════════════════════
def runs(period=P):
return sorted((e["run_no"], e["document_type"], e["task"], e["status"]) for e in db(lambda r: r.data.entries("GRP", period, 9)))
def test_con_20_a_reopened_company_collects_again_later_steps_are_stale_and_the_next_runs_replace_run_1_by_reversal(gl):
exact_world(gl)
ok(api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True}))
eliminate()
run_1 = db(lambda r: r.data.entries("GRP", P, 1))
assert {e["run_no"] for e in run_1} == {1} and next(x for x in state()[0].group_periods if x.period_id == P).status == "eliminated"
# DE01's October is reopened in gl and a 500.00 accrual posted: still open, so it is not collected yet
accrued = [("G-1100", "1060000.00", "0.00"), ("G-2000", "-500.00", "-500.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500000.00", "0.00"),
("G-4000", "-60000.00", "-10000.00"), ("G-6500", "500.00", "500.00")]
gl.set_rows("DE01", P, accrued, status="open")
status, body = api("POST", f"{BASE}/collect-group-data", ACC, {})
assert status == 409 and "company_periods_closed" in str(body)
gl.set_rows("DE01", P, accrued, status="closed") # closed again: the digest is new
out = ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))["result"]
assert (out["run_no"], out["collected"], out["unchanged"]) == (2, ["DE01"], ["UK01"]) # only DE01's new digest is stored
gp = next(x for x in state()[0].group_periods if x.period_id == P)
assert (gp.status, gp.stale_from_step, gp.run_no) == ("collected", "translate", 2) # the period moved back; translation is stale
digests = sorted((c["company_code"], c["run_no"]) for c in db(lambda r: r.data.collections("GRP", P)))
assert digests == [("DE01", 1), ("DE01", 2), ("UK01", 1)]
assert ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))["result"] == {"changed": False} # nothing moved since: nothing written
t2 = ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))["result"]
assert t2["run_no"] == 2 and t2["reversed"] # run 1's translation entry is reversed by a mirror
e2 = ok(api("POST", f"{BASE}/run-eliminations", ACC, {}))["result"]
assert e2["run_no"] == 2 and e2["reversed"] and not set(e2["entries"]) & {e["cons_journal_id"] for e in run_1}
held = {e["cons_journal_id"]: e for e in db(lambda r: r.data.entries("GRP", P, 2))}
for old in run_1: # run 1 is reversed, never deleted
assert held[old["cons_journal_id"]]["status"] == "reversed"
[mirror] = [m for m in held.values() if m["reverses_id"] == old["cons_journal_id"]]
assert [(ln["group_account_id"], -ln["amount"]) for ln in mirror["lines"]] == [(ln["group_account_id"], ln["amount"]) for ln in old["lines"]]
assert {m["run_no"] for m in held.values() if m["status"] == "posted" and not m["reverses_id"]} == {2}
from test_slice_b1 import tb
assert sum(tb().values(), D(0)) == 0 and tb()[("DE01", "G-6500", "", "", "")] == D("430.00") # 500.00 EUR at 0.86 is in the books now
def test_con_20_a_signed_period_is_not_touched_the_change_is_a_difference_until_it_is_reopened(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
before = (runs(), state()[0].group_periods[0].tb_digest)
gl.set_rows("DE01", P, [("G-1100", "1060000.00", "0.00"), ("G-2000", "-500.00", "-500.00"), ("G-3000", "-500000.00", "0.00"),
("G-3100", "-500000.00", "0.00"), ("G-4000", "-60000.00", "-10000.00"), ("G-6500", "500.00", "500.00")])
for path in ("collect-group-data", "recollect-group-data"):
status, body = api("POST", f"{BASE}/{path}", ACC, {})
assert status == 409 and "signed" in str(body)
assert (runs(), state()[0].group_periods[0].tb_digest) == before # figures stay as signed
shown = next(g for g in ok(api("GET", f"{BASE}/readiness", CTRL))["gates"] if g["gate"] == "collected_current")
assert shown["met"] is False and "DE01" in shown["reason"] and "signed figures are unchanged" in shown["reason"]
ok(api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "accrual missed"}))
assert ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))["result"]["run_no"] == 3 # now the change comes in (reopening was run 2)
# ═══ CON-21 materialised statements equal a fresh computation ══════════════════════════════════════════════════════════
def stored():
return {(r["report"], r["node_id"]): (r["amount"], r["prior_year"], r["digest"])
for r in db(lambda rp: rp.data.statements("GRP", P, ["GROUP-IFRS", "GROUP-CF"]))}
def test_con_21_signing_materialises_the_statements_and_doing_it_again_changes_nothing(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
held = stored()
assert held and {r for r, _ in held} == {"GROUP-IFRS"} | ({"GROUP-CF"} if any(r == "GROUP-CF" for r, _ in held) else set())
produced = {r["produced_at"] for r in db(lambda rp: rp.data.statements("GRP", P, ["GROUP-IFRS"]))}
path = f"{BASE}/materialise-statements"
first = ok(api("POST", path, SYSTEM, {}))
second = ok(api("POST", path, SYSTEM, {}))
assert first["changed"] is False and first["written"] == 0 and second == first # equal to a fresh computation at the sign-off digest
assert stored() == held and {r["produced_at"] for r in db(lambda rp: rp.data.statements("GRP", P, ["GROUP-IFRS"]))} == produced
assert first["tb_digest"] == state()[0].group_periods[0].tb_digest
def test_con_21_a_stored_row_that_differs_from_the_fresh_computation_is_put_right_once(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
from sqlalchemy import update
from brick_consolidation.infrastructure.tables import cons_statement
held = stored()
key = next(k for k in held if k[1] != "@meta")
db(lambda rp: rp.conn.execute(update(cons_statement).where(cons_statement.c.report == key[0], cons_statement.c.node_id == key[1]).values(amount=D("1.00"))))
assert stored() != held
fixed = ok(api("POST", f"{BASE}/materialise-statements", SYSTEM, {}))
assert (fixed["changed"], fixed["written"]) == (True, 1) and stored() == held
assert ok(api("POST", f"{BASE}/materialise-statements", SYSTEM, {}))["changed"] is False
def test_con_21_only_a_signed_period_only_the_machine_and_never_over_changed_books(gl):
ready_world(gl)
path = f"{BASE}/materialise-statements"
assert api("POST", path, SYSTEM, {})[0] == 409 # not signed yet
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
for who in (ACC, CTRL, OWNER, AUDITOR):
assert api("POST", path, who, {})[0] == 403 # a machine permission: no person holds it
from sqlalchemy import update
from brick_consolidation.infrastructure.tables import cons_translated_data
db(lambda rp: rp.conn.execute(update(cons_translated_data).where(cons_translated_data.c.group_account_id == "G-1100",
cons_translated_data.c.company_code == "DE01").values(ytd_group=D("1.00"))))
held = stored()
status, body = api("POST", path, SYSTEM, {})
assert status == 409 and body["gate"] == "signed_digest" and stored() == held # the books moved under a signed period: nothing is rewritten
# ═══ CON-18 migration-run sign-off ═══════════════════════════════════════════════════════════════════════════════════
import json # noqa: E402
RUN = "odoo:ACME:2026-11-01"
RUNS = "/cons-migration-runs"
DOORS = json.dumps({"group_id": "GRP", "carried_periods": ["2026-10"], "decisions": {"DE01": "full"}})
def odoo_side(balance="922200.00", structure=None):
return json.dumps([
{"check": "consolidated_balance", "odoo": {"2026-10": {"G-1100": balance}}},
{"check": "structure", "odoo": {"2026-10": structure or {"DE01": "80", "UK01": "100"}}},
{"check": "journals", "odoo": {"2026-10": {"count": 1, "total": "50.00"}}},
{"check": "proportional", "odoo": {"DE01": "80"}}])
def carried_history(gl):
world(gl)
carried_history_rows()
def carried_history_rows():
"""What the loader (CON-34) wrote read-only for the closed period 2026-10: its balance on the report `migration` and one migration journal."""
from brick_consolidation.application import use_cases
with use_cases.repository() as repo:
repo.data.write_statements([{"report": "migration", "group_id": "GRP", "period_id": "2026-10", "node_id": "G-1100", "amount": D("922200.00"),
"prior_year": "", "produced_at": work.now(), "digest": "d"}])
lines = [{"group_account_id": "G-1100", "company_code": "DE01", "partner_company": "", "movement_type": "", "segment": "", "amount": D("50.00")},
{"group_account_id": "G-3000", "company_code": "DE01", "partner_company": "", "movement_type": "", "segment": "", "amount": D("-50.00")}]
repo.data.post_entries([{"cons_journal_id": "9001", "group_id": "GRP", "period_id": "2026-10", "document_type": "migration", "task": "migration",
"run_no": 1, "entry_no": 0, "reverses_id": "", "period_specific": "", "memo": "odoo", "lines": lines}])
def load(checks, run=RUN, doors=DOORS):
return api("POST", RUNS, SYSTEM, {"cons_migration_run_id": run, "doors_json": doors, "checks_json": checks})
def step(name, who, body=None, run=RUN):
return api("POST", f"{RUNS}/{run}/{name}", who, body or {})
def checks_of(run=RUN):
return {c["check"]: c for c in json.loads(ok(api("GET", f"{RUNS}/{run}", OWNER))["checks_json"])}
def test_con_18_checks_are_computed_both_sides_and_the_owner_signs_when_each_is_equal_or_explained(gl):
carried_history(gl)
made = ok(load(odoo_side()))
assert made["status"] == "loaded" and all("ours" not in c for c in checks_of().values())
assert step("reconcile-cons-migration", ACC)[0] == 403 and step("reconcile-cons-migration", OWNER)[0] == 403 # a machine step
ok(step("reconcile-cons-migration", SYSTEM))
got = checks_of()
assert {k: v["status"] for k, v in got.items()} == {"consolidated_balance": "equal", "structure": "equal", "journals": "equal", "proportional": "differs"}
assert got["journals"]["ours"] == {"2026-10": {"count": 1, "total": "50.00"}} and got["proportional"]["ours"] == {"DE01": "full"}
status, body = step("sign-cons-reconciliation", OWNER)
assert status == 409 and "proportional" in str(body) and checks_of()["proportional"].get("explanation") is None # a difference that may be explained, unexplained
assert step("explain-cons-difference", OWNER, {"reason": "r"})[0] == 403 # the accountant explains, the owner signs
assert step("explain-cons-difference", ACC, {"reason": ""})[0] == 409
assert step("explain-cons-difference", ACC, {"reason": "r", "check": "journals"})[0] == 409 # equal: nothing to explain
ok(step("explain-cons-difference", ACC, {"reason": "DE01 was proportional in Odoo; here full with the NCI, decided at the door"}))
assert checks_of()["proportional"]["explanation"]["by"] == "group_accountant@group.example"
assert step("sign-cons-reconciliation", ACC)[0] == 403
signed = ok(step("sign-cons-reconciliation", OWNER))
assert (signed["status"], signed["signed_by"], signed["signed_sub"]) == ("signed", "owner@group.example", "owner@group.example") and signed["signed_at"]
sums = json.loads(signed["checksums"])
assert set(sums) == {"consolidated_balance", "structure", "journals", "proportional", "run"} and all(len(v) == 64 for v in sums.values())
again = ok(step("sign-cons-reconciliation", OWNER))
assert again == signed # a repeat changes nothing
def test_con_18_a_balance_that_differs_cannot_be_explained_only_fixed_and_reloaded(gl):
carried_history(gl)
ok(load(odoo_side(balance="922201.00")))
ok(step("reconcile-cons-migration", SYSTEM))
assert checks_of()["consolidated_balance"]["status"] == "differs"
status, body = step("explain-cons-difference", ACC, {"reason": "a penny"})
assert status == 409 and "must be equal" in str(body)
ok(step("explain-cons-difference", ACC, {"reason": "r", "check": "proportional"}))
status, body = step("sign-cons-reconciliation", OWNER)
assert status == 409 and "consolidated_balance" in str(body)
ok(step("reload-cons-migration", SYSTEM))
assert "ours" not in checks_of()["consolidated_balance"] and "explanation" not in checks_of()["proportional"] # what was computed is dropped
assert step("sign-cons-reconciliation", OWNER)[0] == 409 # not reconciled again yet
def test_con_18_a_signed_run_is_frozen_and_a_load_without_the_doors_is_refused(gl):
carried_history(gl)
assert api("POST", RUNS, SYSTEM, {"cons_migration_run_id": RUN, "checks_json": odoo_side()})[0] == 409 # the doors are not decided
assert load("not json")[0] == 409 and load(json.dumps([{"check": "other", "odoo": 1}]))[0] == 409
ok(load(odoo_side()))
ok(step("reconcile-cons-migration", SYSTEM))
ok(step("explain-cons-difference", ACC, {"reason": "decided at the door"}))
signed = ok(step("sign-cons-reconciliation", OWNER))
assert step("reload-cons-migration", SYSTEM)[0] == 409 and step("discard-cons-migration", OWNER, {"reason": "x"})[0] == 409
assert ok(load(odoo_side(balance="1.00")))["checks_json"] == signed["checks_json"] # a later load never rewrites a signed run
assert ok(api("GET", f"{RUNS}/{RUN}", OWNER)) == signed