dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/test_group_and_ai.py
"""GL-38 what consolidation reads, GL-15 the AI check of a proposed journal, the explanation of a reconciliation break."""
from decimal import Decimal
from conftest import ACCOUNTANT, ACCOUNTANT2, AUDITOR, BOOKKEEPER, CONSOLIDATION, OWNER, SYSTEM, api, must, refused
from fakes import Payments
from world import account_id, books, close_through, invoice, line, post
def group_chart():
must(api("POST", "/charts", ACCOUNTANT, {"chart_id": "GROUP", "kind": "group"}))
for code, typ in (("G-1000", "asset"), ("G-4000", "income"), ("G-6000", "expense"), ("G-3000", "equity"), ("G-2000", "liability")):
must(api("POST", "/accounts", ACCOUNTANT, {"chart_id": "GROUP", "code": code, "name": code, "type": typ}))
def gid(code):
from erp_platform.seam import derived_id
return derived_id("account", "GROUP", code)
def test_a_group_account_cannot_be_posted_to_and_the_mapping_is_dated_with_history():
books()
group_chart()
s, b = post("X-1", "x", "2026-10-15", [{"account_id": str(gid("G-4000")), "credit": "5.00"}, line(purpose="receivable", debit="5.00")])
refused((s, b), 422, "[group_chart_not_postable]")
must(api("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": str(account_id("4000")), "group_account_id": str(gid("G-4000")), "valid_from": "2026-01-01"}))
refused(api("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": str(account_id("4000")), "group_account_id": str(gid("G-4000")),
"valid_from": "2026-10-22"}), 422, "tomorrow")
must(api("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": str(account_id("4000")), "group_account_id": str(gid("G-6000")), "valid_from": "2026-11-01"}))
hist = must(api("GET", f"/accounts/{account_id('4000')}", ACCOUNTANT))["group_account_history"]
assert [h["valid_from"] for h in hist] == ["2026-01-01", "2026-11-01"]
def test_consolidation_reads_group_balances_with_a_stable_digest_and_the_unmapped_listed():
books()
group_chart()
invoice("INV-G1", on="2026-10-10")
post("COST-G1", "x", "2026-10-12", [line(account="6100", debit="300.00"), line(account="2100", credit="300.00")], source_module="inventory")
for code, g in (("4000", "G-4000"), ("1100", "G-1000"), ("6100", "G-6000")):
must(api("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": str(account_id(code)), "group_account_id": str(gid(g)), "valid_from": "2026-01-01"}))
r = must(api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "UK01", "ledger_id": "LEAD", "period_id": "2026-10"}))
assert r["period_status"] == "open" and r["functional_currency"] == "GBP" and r["group_chart_id"] == "GROUP"
by = {str(x["group_account_id"]): x for x in r["rows"]}
assert by[str(gid("G-4000"))]["ytd_amount"] == "-1000.00" and by[str(gid("G-1000"))]["period_amount"] == "1200.00"
assert {u["code"] for u in r["unmapped"]} == {"2100", "2200"}
again = must(api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "UK01", "ledger_id": "LEAD", "period_id": "2026-10"}))
assert again["digest"] == r["digest"]
invoice("INV-G2", on="2026-10-11", net="10.00", vat="2.00")
assert must(api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "UK01", "ledger_id": "LEAD", "period_id": "2026-10"}))["digest"] != r["digest"]
gate = must(api("GET", "/readiness/group-mapped", ACCOUNTANT, {}))["gate"]
assert gate["met"] is False and "2100" in gate["reason"]
def test_the_group_extract_includes_the_opening_and_is_refused_to_everyone_but_consolidation():
books()
group_chart()
invoice("INV-G1", on="2026-03-10")
for code, g in (("4000", "G-4000"), ("1100", "G-1000"), ("2200", "G-2000"), ("3000", "G-3000")):
must(api("POST", "/accounts/group-mapping", ACCOUNTANT, {"account_id": str(account_id(code)), "group_account_id": str(gid(g)), "valid_from": "2026-01-01"}))
close_through("2026-12")
must(api("POST", "/fiscal-years/FY2026/close-year", ACCOUNTANT, {}))
r = must(api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "UK01", "ledger_id": "LEAD", "period_id": "2027-01"}))
opening = [x for x in r["rows"] if x["movement_type"] == "opening"]
assert {str(x["group_account_id"]): x["ytd_amount"] for x in opening} == {str(gid("G-1000")): "1200.00", str(gid("G-2000")): "-200.00", str(gid("G-3000")): "-1000.00"}
for who in (ACCOUNTANT, AUDITOR, SYSTEM, BOOKKEEPER):
if who is SYSTEM:
continue
assert api("GET", "/group-reporting/balances", who, {"company_code": "UK01", "ledger_id": "LEAD", "period_id": "2027-01"})[0] == 403
refused(api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "UK01", "ledger_id": "NOPE", "period_id": "2027-01"}), 422, "[ledger_assigned]")
assert api("GET", "/group-reporting/balances", CONSOLIDATION, {"company_code": "ZZ", "ledger_id": "LEAD", "period_id": "2027-01"})[0] == 404
# ── GL-15 ────────────────────────────────────────────────────────────────────────────────────────────────────────────
def history_of_accruals(n=5):
for i in range(n):
must(api("POST", "/journals/propose-manual", ACCOUNTANT, {"journal_date": f"2026-0{i + 1}-20", "memo": "accrual",
"lines": [line(account="6100", debit=f"{100 + i}.00"), line(account="2100", credit=f"{100 + i}.00")]}))
rows = must(api("GET", "/journals", ACCOUNTANT, {"status": "proposed"}))["journals"]
for r in rows:
must(api("POST", f"/journals/{r['journal_id']}/approve-manual", ACCOUNTANT2, {}))
def test_the_ai_check_lists_findings_with_evidence_and_never_blocks():
books()
history_of_accruals()
p = must(api("POST", "/journals/propose-manual", ACCOUNTANT, {"journal_date": "2026-10-20",
"lines": [line(account="6100", debit="102.00"), line(account="2100", credit="102.00")]}))
j = must(api("GET", f"/journals/{p['journal_id']}", ACCOUNTANT))
out = must(api("POST", f"/journals/{p['journal_id']}/ai-check-journal", SYSTEM, {}))
kinds = {f["finding"]: f for f in out["findings"]}
assert "duplicate" in kinds and "same lines and amount as MISC/FY2026/" in kinds["duplicate"]["evidence"] and j["status"] == "proposed"
odd = must(api("POST", "/journals/propose-manual", ACCOUNTANT, {"journal_date": "2026-10-20", "lines": [line(account="6400", debit="9.00"), line(account="2200", credit="9.00")]}))
kinds2 = {f["finding"] for f in must(api("POST", f"/journals/{odd['journal_id']}/ai-check-journal", SYSTEM, {}))["findings"]}
assert "odd_pair" in kinds2
big = must(api("POST", "/journals/propose-manual", ACCOUNTANT, {"journal_date": "2026-10-20", "lines": [line(account="6100", debit="5000.00"), line(account="2100", credit="5000.00")]}))
assert "amount_out_of_pattern" in {f["finding"] for f in must(api("POST", f"/journals/{big['journal_id']}/ai-check-journal", SYSTEM, {}))["findings"]}
# the approver decides each; the journal is unchanged; approval does not wait
fid = out["findings"][0]["finding_id"]
done = must(api("POST", f"/journals/{p['journal_id']}/accept-journal-ai-findings", ACCOUNTANT2, {"decisions": [{"finding_id": fid, "decision": "dismissed"}]}))
d = next(f for f in done["findings"] if f["finding_id"] == fid)
assert d["decision"] == "dismissed" and d["decided_by"] == "accountant2@firm.example"
assert must(api("POST", f"/journals/{p['journal_id']}/approve-manual", ACCOUNTANT2, {}))["status"] == "posted"
assert api("POST", f"/journals/{p['journal_id']}/ai-check-journal", ACCOUNTANT, {})[0] == 403
assert api("POST", f"/journals/{p['journal_id']}/accept-journal-ai-findings", BOOKKEEPER, {"decisions": []})[0] in (403, 422)
def test_findings_appear_on_the_proposal_for_the_approver_and_the_step_can_be_switched_off():
books()
history_of_accruals()
p = must(api("POST", "/journals/propose-manual", ACCOUNTANT, {"journal_date": "2026-10-20", "lines": [line(account="6100", debit="100.00"), line(account="2100", credit="100.00")]}))
from sqlalchemy import text # noqa: F401
must(api("POST", "/controls", OWNER, {"key": "ai.ai_check_journal.mode", "value": "off"}))
from conftest import on_day
with on_day("2026-10-23T09:00:00+00:00"):
refused(api("POST", f"/journals/{p['journal_id']}/ai-check-journal", SYSTEM, {}), 409, "off")
_ = ACCOUNTANT2
def test_a_break_comes_with_a_likely_cause_and_posts_nothing():
books()
pay = Payments().install()
post("pay_x", "receipt", "2026-10-20", [line(account="1200", debit="40.00"), line(purpose="receivable", credit="40.00")], source_module="payments")
pay.invoices = {}
r = must(api("POST", "/reconcile/receivable", BOOKKEEPER, {"as_at": "2026-10-31"}))
n = len(must(api("GET", "/journals", ACCOUNTANT))["journals"])
exp = must(api("POST", "/reconciliations/explain-break", BOOKKEEPER, {"recon_id": r["recon_id"], "document": "unattributed:pay_x"}))
assert "without the invoice it pays" in exp["cause"] and "nothing was posted" in exp["note"]
assert len(must(api("GET", "/journals", ACCOUNTANT))["journals"]) == n
_ = Decimal