dodil/erp-brick-consolidation
DODIL ERP brick: consolidation
tests/test_slice_b2.py
"""Slice B2: CON-11 group statements, CON-12 uploads for companies outside gl, CON-13 readiness, CON-14 sign-off and reopen."""
import hashlib
from decimal import Decimal
from conftest import ALL, api, as_role, person, tool
from erp_platform.after import join_after_response
from test_slice_b1 import add_rules, eliminate, exact_world
from world import ACC, CTRL, CTRL_SAME_EMAIL, CTRL_SAME_SUB, P, db, ok, state, world
from brick_consolidation.application import reports
from brick_consolidation.domain import consolidation as C
D = Decimal
BASE = f"/groups/GRP/group-periods/{P}"
CTRL2 = as_role("group_controller", email="second.controller@group.example")
def node(i, parent, seq, label, sign=1, klass="non_cash"):
return {"node_id": i, "parent_node_id": parent, "seq": seq, "label": label, "sign": sign, "cash_flow_class": klass}
def assign(a, b, node_id, basis="balance", dc="always", other=None):
return {"from_code": a, "to_code": b, "node_id": node_id, "dc_rule": dc, "other_node_id": other, "basis": basis}
def publish_versions(gl):
codes = ["G-1100", "G-1300", "G-1400", "G-1450", "G-1600", "G-1700", "G-1990", "G-2000", "G-2300", "G-3000", "G-3100", "G-3900", "G-3950",
"G-3960", "G-4000", "G-4100", "G-4300", "G-5000", "G-5100", "G-6500", "G-6600", "G-6900"]
gl.chart = {c: c for c in codes}
gl.versions["GROUP-IFRS"] = {"kind": "balance_sheet_pl", "version_no": 3, "nodes": [
node("BS-A", None, 1, "Assets"), node("A-CASH", "BS-A", 1, "Cash"), node("A-GW", "BS-A", 2, "Goodwill"), node("A-OTHER", "BS-A", 3, "Other assets"),
node("BS-L", None, 2, "Liabilities", -1), node("L-PAY", "BS-L", 1, "Payables", -1),
node("BS-E", None, 3, "Equity", -1), node("E-CAP", "BS-E", 1, "Share capital and reserves", -1), node("E-RES", "BS-E", 2, "Translation and minority", -1),
node("PL", None, 4, "Result", -1), node("PL-REV", "PL", 1, "Revenue", -1), node("PL-EXP", "PL", 2, "Expenses", -1)],
"assignments": [assign("G-1100", "G-1100", "A-CASH"), assign("G-1700", "G-1700", "A-GW"), assign("G-1300", "G-1699", "A-OTHER"),
assign("G-1990", "G-1999", "A-OTHER"), assign("G-2000", "G-2999", "L-PAY"), assign("G-3000", "G-3199", "E-CAP"),
assign("G-3900", "G-3999", "E-RES"), assign("G-4000", "G-4999", "PL-REV"), assign("G-5000", "G-6999", "PL-EXP")]}
gl.versions["GROUP-CF"] = {"kind": "cash_flow", "version_no": 1, "nodes": [
node("CF-OP", None, 1, "Operating", -1, "operating"), node("CF-INV", None, 2, "Investing", -1, "investing"),
node("CF-FIN", None, 3, "Financing", -1, "financing"), node("CF-CASH", None, 4, "Cash", 1, "cash"), node("CF-NC", None, 5, "Translation reserve", -1, "non_cash")],
"assignments": [assign("G-4000", "G-6999", "CF-OP", "movement"), assign("G-2000", "G-2999", "CF-OP", "movement"),
assign("G-1600", "G-1799", "CF-INV", "movement"), assign("G-3000", "G-3199", "CF-FIN", "movement"),
assign("G-3950", "G-3999", "CF-FIN", "movement"), assign("G-1100", "G-1100", "CF-CASH", "movement"),
assign("G-3900", "G-3900", "CF-NC", "movement")]}
ok(api("POST", "/groups", ACC, {"group_id": "GRP", "fsv_id": "GROUP-IFRS", "cash_flow_fsv_id": "GROUP-CF", "version": state()[1]}))
def ready_world(gl):
exact_world(gl)
publish_versions(gl)
return eliminate()
def by_node(report):
return {line["node_id"]: line for line in report["lines"]}
def report(headers=ACC, **q):
return ok(api("GET", f"{BASE}/report", headers, q))
# ── CON-11 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def test_con_11_balance_sheet_and_result_through_gls_version_balance_and_open_to_companies_and_entries(gl):
ready_world(gl)
rep = report(fsv_id="GROUP-IFRS")
n = by_node(rep)
assert (rep["fsv_id"], rep["version_no"], rep["ledger_id"], rep["run_no"], rep["materialised"]) == ("GROUP-IFRS", 3, "IFRS", 1, False) # names version, ledger, run
assert D(n["A-CASH"]["amount"]) == D("922200.00") and D(n["A-GW"]["amount"]) == D("87000.00") and D(n["BS-A"]["amount"]) == D("1009200.00")
assert D(n["E-CAP"]["amount"]) == D("774000.00") and D(n["E-RES"]["amount"]) == D("183600.00") and D(n["PL-REV"]["amount"]) == D("51600.00")
assert D(n["BS-A"]["amount"]) == D(n["BS-E"]["amount"]) + D(n["PL"]["amount"]) + D(n["BS-L"]["amount"]) # the group balances to the penny
assert rep["unassigned"] == [] and D(n["A-CASH"]["by_company"]["DE01"]) == D("922200.00") and D(n["E-CAP"]["by_company"]["UK01"]) == D("774000.00")
[behind] = n["A-GW"]["entries"] # the goodwill opens to its entry
assert behind["document_type"] == "investment" and D(behind["amount"]) == D("87000.00") and behind["entry_no"] > 0
assert "GL-04" in rep["drill_down"]["gl"]
# the same through the agent's door, and a version gl has not published is refused
assert tool(ACC, "run_group_report", {"group_id": "GRP", "period_id": P, "fsv_id": "GROUP-IFRS"})[0] == "ok"
status, text = api("GET", f"{BASE}/report", ACC, {"fsv_id": "NOPE"})
assert status == 409 and "NOPE" in str(text)
assert api("GET", f"{BASE}/report", as_role("auditor"), {"fsv_id": "GROUP-IFRS"})[0] == 200
def test_con_11_by_segment_and_beside_the_prior_year(gl):
ready_world(gl)
db(lambda r: r.data.write_statements([{"report": "GROUP-IFRS", "group_id": "GRP", "period_id": "2025-10", "node_id": "A-CASH", "amount": D("100.00"),
"prior_year": "", "produced_at": None, "digest": "x"}]))
n = by_node(report(fsv_id="GROUP-IFRS"))
assert D(n["A-CASH"]["prior_year"]) == D("100.00") and n["A-GW"]["prior_year"] is None
assert {k: D(v) for k, v in n["A-CASH"]["by_segment"].items()} == {"": D("922200.00")}
def test_con_11_cash_flow_shows_the_translation_difference_on_cash_as_its_own_line(gl):
ready_world(gl)
rep = report(fsv_id="GROUP-CF")
cf = {k: D(v) for k, v in rep["cash_flow"].items()}
assert cf["cash"] == D("922200.00") and cf["operating"] == D("51600.00") and cf["investing"] == D("-87000.00")
assert cf["financing"] == D("948120.00")
assert cf["translation_difference_on_cash"] == D("9480.00") # IAS 7.28: what the translation reserve moved, on its own
assert cf["operating"] + cf["investing"] + cf["financing"] + cf["translation_difference_on_cash"] == cf["cash"]
def test_con_11_changes_in_equity_read_the_movement_types(gl):
ready_world(gl)
add_rules(("G-6600", "average"))
j = ok(api("POST", "/cons-journals/propose-cons-entry", ACC, {"cons_journal_id": "CAP", "group_id": "GRP", "period_id": P, "lines": [
{"group_account_id": "G-3000", "company_code": "UK01", "amount": "-5.00", "movement_type": "capital_change"},
{"group_account_id": "G-1100", "company_code": "UK01", "amount": "5.00"}]}))
ok(api("POST", f"/cons-journals/{j['cons_journal_id']}/approve-cons-entry", CTRL, {}))
rep = ok(api("GET", f"{BASE}/report", ACC, {"report": "equity_changes"}))
assert rep["movements"]["capital_change"] == "-5.00" and D(rep["closing"]) == D("-774005.00") and rep["by_company"]["UK01"]["capital_change"] == "-5.00"
def test_con_11_a_signed_period_is_materialised_and_stable(gl):
ready_world(gl)
before = report(fsv_id="GROUP-IFRS")
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
rows = db(lambda r: r.data.statements("GRP", P, ["GROUP-IFRS", "GROUP-CF"]))
assert {r["report"] for r in rows} == {"GROUP-IFRS", "GROUP-CF"} and len({r["digest"] for r in rows}) == 1
after = report(fsv_id="GROUP-IFRS")
assert after["materialised"] is True and after["produced_at"]
assert {k: v["amount"] for k, v in by_node(after).items()} == {k: v["amount"] for k, v in by_node(before).items()}
assert (after["fsv_id"], after["version_no"], after["ledger_id"], after["run_no"]) == ("GROUP-IFRS", 3, "IFRS", 1)
fresh = db(lambda r: reports.compute(r, state()[0], next(x for x in state()[0].group_periods if x.period_id == P), "GROUP-IFRS",
*reports.consolidated(r, state()[0], next(x for x in state()[0].group_periods if x.period_id == P))))
assert {k: D(v["amount"]) for k, v in by_node(after).items()} == {k: v["amount"] for k, v in by_node(fresh).items()} # equals a fresh computation at the digest
gl.versions["GROUP-IFRS"]["version_no"] = 4 # gl publishes a new version: the signed report does not move
assert report(fsv_id="GROUP-IFRS")["version_no"] == 3
# ── CON-12 ─────────────────────────────────────────────────────────────────────────────────────────────────────
CSV = "group_account_id,partner_company,movement_type,segment,ytd_amount\nG-1100,,,,1000.00\nG-3000,,capital_change,,-1000.00\n"
def us01_world(gl):
"""US01 is kept in another package: a structure row for it, no gl company."""
from brick_consolidation.application import use_cases
from brick_consolidation.domain.group import GroupStructure
world(gl, acquired=False, opened=False)
gl.chart = {c: c for c in ("G-1100", "G-2000", "G-3000", "G-3100")}
gl.set_rows("UK01", P, [("G-1100", "500.00", "0.00"), ("G-3000", "-500.00", "0.00")])
with use_cases.repository() as repo:
g, v = repo.groups.get("GRP")
g.group_structures.append(GroupStructure(group_id="GRP", company_code="US01", valid_from_period="2026-04", investor_company_code="UK01",
ownership_pct=D("100"), voting_pct=D("100"), method="full", change_id="elsewhere"))
repo.groups.save(g, v)
ok(api("POST", "/groups/GRP/group-periods", ACC, {"period_id": P}))
for kind, key, rate in (("closing", "2026-10-31", "0.800000"),):
gl.rates[("USD", kind, key)] = D(rate)
for m in ("2026-04", "2026-05", "2026-06", "2026-07", "2026-08", "2026-09", "2026-10"):
gl.rates[("USD", "average", m)] = D("0.790000")
def upload(csv=CSV, headers=ACC, company="US01", currency="USD"):
return api("POST", f"{BASE}/upload-reported-data", headers, {"company_code": company, "currency": currency, "csv": csv})
def test_con_12_upload_validates_stores_the_file_with_its_sha_and_waits_for_a_second_person(gl, store):
us01_world(gl)
out = ok(upload())["result"]
join_after_response()
sha = hashlib.sha256(CSV.encode()).hexdigest()
assert out["sha256"] == sha and out["object_key"].startswith("raw/consolidation/GRP/2026-10/US01-1-") and out["rows"] == 2
assert store.put_[out["object_key"]] == (CSV.encode(), sha) # the file is kept under raw/consolidation/<group>/<period>/
[row] = db(lambda r: r.data.uploads("GRP", P, "US01"))
assert row["status"] == "submitted" and row["submitted_by"] == "group_accountant@group.example" and row["sha256"] == sha
assert db(lambda r: r.data.reported("GRP", [P], ["US01"])) == [] # not reported data until accepted
# a second upload replaces the one still waiting
ok(upload(CSV.replace("1000.00", "2000.00").replace("-1000.00", "-2000.00")))
assert [u["status"] for u in db(lambda r: r.data.uploads("GRP", P, "US01"))] == ["replaced", "submitted"]
def test_con_12_bad_files_are_refused_with_the_line(gl):
us01_world(gl)
head = "group_account_id,partner_company,movement_type,segment,ytd_amount\n"
cases = {"unknown account": head + "G-9999,,,,5.00\nG-3000,,capital_change,,-5.00\n", "does not net": head + "G-1100,,,,5.00\nG-3000,,capital_change,,-4.00\n",
"cents": head + "G-1100,,,,5.001\nG-3000,,capital_change,,-5.001\n", "partner": head + "G-1100,XX99,,,5.00\nG-3000,,capital_change,,-5.00\n",
"twice": head + "G-1100,,,,5.00\nG-1100,,,,-5.00\n", "no rows": head, "wrong header": "a,b\n1,2\n",
"overflow": head + "G-1100,,,,12345678901234567.00\nG-3000,,capital_change,,-12345678901234567.00\n"}
for name, csv in cases.items():
status, text = upload(csv)
assert status in (409, 422), (name, status, text)
assert db(lambda r: r.data.uploads("GRP", P)) == []
status, text = upload(company="UK01") # a company in gl cannot be uploaded: its data comes from gl
assert status == 409 and "kept in gl" in str(text) and "not_kept_in_gl" in str(text)
assert upload(company="DE99")[0] == 409 # not in scope
assert upload(currency="usd")[0] == 422
assert upload(headers=CTRL)[0] == 403 # the controller does not upload
def test_con_12_the_uploader_cannot_accept_it_on_either_identity_and_acceptance_makes_it_the_companys_data(gl):
us01_world(gl)
ok(upload(headers=person("loader@group.example", ALL)))
accept = {"company_code": "US01", "currency": "USD"}
same_email = person("loader@group.example", ALL)
other_email_same_sub = as_role("group_controller", email="elsewhere@group.example", sub="loader@group.example")
for who in (same_email, other_email_same_sub):
status, body = api("POST", f"{BASE}/accept-reported-upload", who, accept)
assert status == 409 and body["gate"] == "second_person", (status, body)
assert api("POST", f"{BASE}/accept-reported-upload", ACC, accept)[0] == 403
ok(api("POST", f"{BASE}/accept-reported-upload", CTRL, accept))
rows = db(lambda r: r.data.reported("GRP", [P], ["US01"]))
assert {(r["group_account_id"], r["movement_type"], r["ytd_local"], r["source"], r["local_currency"]) for r in rows} == {
("G-1100", "", D("1000.00"), "upload", "USD"), ("G-3000", "capital_change", D("-1000.00"), "upload", "USD")}
assert [u["status"] for u in db(lambda r: r.data.uploads("GRP", P, "US01"))] == ["accepted"]
assert ok(api("POST", f"{BASE}/accept-reported-upload", CTRL, accept))["result"] == {"changed": False}
def test_con_12_collect_takes_the_accepted_upload_and_translation_flows_on(gl):
us01_world(gl)
status, text = api("POST", f"{BASE}/collect-group-data", ACC, {})
assert status == 409 and "upload" in str(text) # a company outside gl with nothing accepted blocks the collection
ok(upload())
ok(api("POST", f"{BASE}/accept-reported-upload", CTRL, {"company_code": "US01", "currency": "USD"}))
out = ok(api("POST", f"{BASE}/collect-group-data", ACC, {}))["result"]
assert out["collected"] == ["UK01"] and "US01" in out["unchanged"]
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
t = {(r["company_code"], r["group_account_id"]): r for r in db(lambda r: r.data.translated("GRP", P, 1))}
assert t[("US01", "G-1100")]["ytd_group"] == D("800.00") # 1,000.00 USD at the closing rate
assert [c["source"] for c in db(lambda r: r.data.collections("GRP", P)) if c["company_code"] == "US01"] == ["upload"]
def test_con_12_a_replacement_after_collection_marks_later_steps_stale(gl):
us01_world(gl)
ok(upload())
ok(api("POST", f"{BASE}/accept-reported-upload", CTRL, {"company_code": "US01", "currency": "USD"}))
for step in ("collect-group-data", "translate-currencies"):
ok(api("POST", f"{BASE}/{step}", ACC, {}))
ok(upload(CSV.replace("1000.00", "1500.00")))
assert db(lambda r: r.data.uploads("GRP", P, "US01"))[-1]["status"] == "submitted"
ok(api("POST", f"{BASE}/accept-reported-upload", CTRL, {"company_code": "US01", "currency": "USD"}))
gp = next(x for x in state()[0].group_periods if x.period_id == P)
assert (gp.status, gp.run_no, gp.stale_from_step) == ("collected", 2, "translate")
assert [u["status"] for u in db(lambda r: r.data.uploads("GRP", P, "US01"))] == ["replaced", "accepted"]
ok(api("POST", f"{BASE}/translate-currencies", ACC, {})) # the next run takes run_no + 1
assert {r["ytd_group"] for r in db(lambda r: r.data.translated("GRP", P, 2)) if r["company_code"] == "US01" and r["group_account_id"] == "G-1100"} == {D("1200.00")}
# ── CON-13 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def readiness(headers=CTRL):
return ok(api("GET", f"{BASE}/readiness", headers))
def gates(r):
return {g["gate"]: g for g in r["gates"]}
def test_con_13_every_gate_is_listed_with_met_reason_and_fix_and_a_clean_close_is_ready(gl):
ready_world(gl)
r = readiness()
assert r["met"] is True and [g["gate"] for g in r["gates"]] == [
"collected_current", "carry_forward_current", "account_roles_complete", "tax_rate_known", "no_step_stale", "reported_nets_zero", "translated_nets_zero", "intercompany_within_tolerance", "investments_eliminated",
"no_manual_entry_waiting", "consolidated_tb_nets_zero", "balance_sheet_balances"]
assert all(set(g) == {"gate", "met", "reason", "fix"} and g["met"] for g in r["gates"])
assert tool(CTRL, "group_close_readiness", {"group_id": "GRP", "period_id": P})[0] == "ok"
assert api("GET", f"{BASE}/readiness", as_role("auditor"))[0] == 200
def test_con_13_each_failing_gate_names_the_company_pair_or_account(gl):
ready_world(gl)
ok(api("POST", "/cons-journals/propose-cons-entry", ACC, {"cons_journal_id": "IC", "group_id": "GRP", "period_id": P, "lines": [
{"group_account_id": "G-1300", "company_code": "UK01", "partner_company": "DE01", "amount": "100.00"},
{"group_account_id": "G-1990", "company_code": "DE01", "amount": "-100.00"}]}))
r = readiness()
assert not r["met"] and "IC" not in gates(r)["no_manual_entry_waiting"]["reason"] and "waits for approval" in gates(r)["no_manual_entry_waiting"]["reason"]
assert gates(r)["no_manual_entry_waiting"]["fix"] and not gates(r)["no_manual_entry_waiting"]["met"]
jid = db(lambda rp: rp.data.waiting("GRP", P))[0]
ok(api("POST", f"/cons-journals/{jid}/approve-cons-entry", CTRL, {}))
g = gates(readiness())
assert g["no_manual_entry_waiting"]["met"] and not g["intercompany_within_tolerance"]["met"] and "UK01/DE01" in g["intercompany_within_tolerance"]["reason"]
# a company's figures that do not net to zero, and gl's figures moving after the collection
from sqlalchemy import update
from brick_consolidation.infrastructure.tables import cons_reported_data
db(lambda rp: rp.conn.execute(update(cons_reported_data).where(cons_reported_data.c.company_code == "DE01",
cons_reported_data.c.group_account_id == "G-1100").values(ytd_local=D("1060001.00"))))
gl.set_rows("DE01", P, [("G-1100", "1060500.00", "0.00"), ("G-3000", "-500000.00", "0.00"), ("G-3100", "-500500.00", "0.00"), ("G-4000", "-60000.00", "-10000.00")])
g = gates(readiness())
assert not g["reported_nets_zero"]["met"] and "DE01 nets to 1.00" in g["reported_nets_zero"]["reason"]
assert not g["collected_current"]["met"] and "DE01" in g["collected_current"]["reason"] and "UK01" not in g["collected_current"]["reason"]
assert g["translated_nets_zero"]["met"] # the translated rows themselves were not touched
def test_con_13_before_the_eliminations_the_investment_and_the_step_are_named(gl):
exact_world(gl)
g = gates(readiness())
assert not g["no_step_stale"]["met"] and "translated" in g["no_step_stale"]["reason"]
assert not g["investments_eliminated"]["met"] and "G-1600" in g["investments_eliminated"]["reason"] and "DE01" in g["investments_eliminated"]["reason"]
assert api("GET", f"{BASE}/readiness", ACC)[0] == 200 and api("GET", "/groups/GRP/group-periods/2030-01/readiness", ACC)[0] == 409
# ── CON-14 ─────────────────────────────────────────────────────────────────────────────────────────────────────
def signoffs():
return ok(api("GET", f"{BASE}/signoffs", CTRL))["signoffs"]
def test_con_14_the_preparer_cannot_sign_under_any_identity_then_a_second_controller_signs(gl):
ready_world(gl)
same_both = person("group_accountant@group.example", ALL)
for who in (same_both, CTRL_SAME_EMAIL, CTRL_SAME_SUB):
status, body = api("POST", f"{BASE}/sign-off-group-close", who, {})
assert status == 409 and body["gate"] == "second_person", (status, body)
assert api("POST", f"{BASE}/sign-off-group-close", ACC, {})[0] == 403
gp = next(x for x in state()[0].group_periods if x.period_id == P)
assert gp.status == "eliminated" and signoffs() == []
out = ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))["result"]
gp = next(x for x in state()[0].group_periods if x.period_id == P)
assert gp.status == "signed" and gp.signed_by == "group_controller@group.example" and gp.signed_at and gp.tb_digest == out["tb_digest"]
tb, _ = db(lambda r: reports.consolidated(r, state()[0], gp))
assert gp.tb_digest == C.tb_digest(tb) and len(gp.tb_digest) == 64
[row] = signoffs()
assert (row["action"], row["by_email"], row["tb_digest"], row["run_no"]) == ("signed", "group_controller@group.example", gp.tb_digest, 1)
version = state()[1]
assert ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))["result"] == {"changed": False} and state()[1] == version # a repeat changes nothing
def test_con_14_a_signed_period_refuses_every_run_until_reopened(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
for step in ("collect-group-data", "recollect-group-data", "translate-currencies", "run-eliminations"):
status, text = api("POST", f"{BASE}/{step}", ACC, {})
assert status == 409, (step, status, text)
assert api("POST", f"{BASE}/ic-inventory", ACC, {"holder_company": "DE01", "seller_company": "UK01", "none_stated": True})[0] == 409
assert api("POST", "/cons-journals/propose-cons-entry", ACC, {"cons_journal_id": "L", "group_id": "GRP", "period_id": P, "lines": [
{"group_account_id": "G-1100", "company_code": "UK01", "amount": "1.00"}, {"group_account_id": "G-1990", "company_code": "UK01", "amount": "-1.00"}]})[0] == 409
assert api("POST", f"{BASE}/upload-reported-data", ACC, {"company_code": "DE01", "currency": "EUR", "csv": CSV})[0] == 409
def test_con_14_readiness_must_be_met_to_sign(gl):
ready_world(gl)
ok(api("POST", "/cons-journals/propose-cons-entry", ACC, {"cons_journal_id": "W", "group_id": "GRP", "period_id": P, "lines": [
{"group_account_id": "G-1100", "company_code": "UK01", "amount": "1.00"}, {"group_account_id": "G-1990", "company_code": "UK01", "amount": "-1.00"}]}))
status, body = api("POST", f"{BASE}/sign-off-group-close", CTRL, {})
assert status == 409 and body["gate"] == "no_manual_entry_waiting" and "waits for approval" in body["detail"] and body["fix"]
assert signoffs() == []
def test_con_14_a_late_gl_posting_shows_as_a_changed_digest_and_the_signed_figures_stay_until_reopened_by_another_controller(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
signed = {k: v["amount"] for k, v in by_node(report(fsv_id="GROUP-IFRS")).items()}
first = signoffs()[0]
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")]) # DE01 reopens in gl and posts
late = gates(readiness())["collected_current"]
assert not late["met"] and "DE01" in late["reason"] and "signed figures are unchanged" in late["reason"]
assert {k: v["amount"] for k, v in by_node(report(fsv_id="GROUP-IFRS")).items()} == signed
status, body = api("POST", f"{BASE}/recollect-group-data", ACC, {})
assert status == 409 and "signed" in str(body) # the change waits for the reopening
assert api("POST", f"{BASE}/reopen-group-period", CTRL, {"reason": "late posting"})[0] == 409 # not by the signer
status, body = api("POST", f"{BASE}/reopen-group-period", CTRL, {"reason": "late posting"})
assert body["gate"] == "second_person"
assert api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": ""})[0] == 422
assert api("POST", f"{BASE}/reopen-group-period", as_role("group_accountant"), {"reason": "x"})[0] == 403
out = ok(api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "late posting in DE01"}))["result"]
assert out["provisional"] is True and out["run_no"] == 2
gp = next(x for x in state()[0].group_periods if x.period_id == P)
assert (gp.status, gp.reopened_reason, gp.run_no) == ("collected", "late posting in DE01", 2)
history = signoffs()
assert [h["action"] for h in history] == ["signed", "reopened"] and history[0] == first # signed history is never rewritten
assert (history[1]["by_email"], history[1]["reason"], history[1]["tb_digest"]) == ("second.controller@group.example", "late posting in DE01", first["tb_digest"])
assert ok(api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "late posting in DE01"}))["result"] == {"changed": False}
# the figures are provisional: re-run, and the second signing records the new digest
ok(api("POST", f"{BASE}/recollect-group-data", ACC, {}))
ok(api("POST", f"{BASE}/translate-currencies", ACC, {}))
eliminate()
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL2, {}))
again = signoffs()
assert [h["action"] for h in again] == ["signed", "reopened", "signed"] and again[2]["tb_digest"] != first["tb_digest"] and again[0] == first
assert {k: v["amount"] for k, v in by_node(report(fsv_id="GROUP-IFRS")).items()} != signed
def test_con_14_a_period_followed_by_a_signed_one_is_not_reopened(gl):
ready_world(gl)
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
from brick_consolidation.application import use_cases
from brick_consolidation.domain.group import GroupPeriod
with use_cases.repository() as repo:
g, v = repo.groups.get("GRP")
g.group_periods.append(GroupPeriod(group_id="GRP", period_id="2026-11", run_no=1, status="signed", signed_by="x", signed_sub="x"))
repo.groups.save(g, v)
status, body = api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "x"})
assert status == 409 and body["gate"] == "next_period_signed"
def test_con_14_a_sign_off_racing_a_recollect_collides_on_the_one_claim_key(gl):
ready_world(gl)
loaded, v = state()
ok(api("POST", f"{BASE}/sign-off-group-close", CTRL, {}))
assert state()[1] == v + 1 # sign-off claimed the group's next version
status, text = api("POST", f"{BASE}/recollect-group-data", ACC, {"version": v}) # a recollect that read the group before the sign-off
assert status == 409 and "changed since you loaded" in str(text)
import pytest
from brick_consolidation.application import use_cases
with pytest.raises(Exception) as late, use_cases.repository() as repo:
repo.groups.save(loaded, v)
assert "changed since" in str(late.value) or "Conflict" in type(late.value).__name__
ok(api("POST", f"{BASE}/reopen-group-period", CTRL2, {"reason": "r"}))
assert state()[1] == v + 2 # and so does the reopen
def test_con_14_second_person_can_be_switched_off(gl):
ready_world(gl)
ok(api("POST", "/controls/signoff_second_person", as_role("owner"), {"value": False, "valid_from": "2026-11-20"}))
assert ok(api("POST", f"{BASE}/sign-off-group-close", person("group_accountant@group.example", ALL), {}))["result"]["changed"] is True