dodil/erp-brick-gl
DODIL ERP brick: general ledger
tests/test_intercompany.py
"""GL-51 a cost charged between two companies, GL-52 matching the pair."""
from decimal import Decimal
from conftest import ACCOUNTANT, ACCOUNTANT2, OWNER, api, must, person, refused
from erp_platform import seam
from fastapi import FastAPI
from world import account_id, books, close_through
DE_BK = person("de.bk@firm.example", ["gl:journal:read", "gl:journal:write", "gl:company:read"], sub="sub-debk")
def two_companies(tolerance=None):
books(extra_map={"ic_receivable": "1100", "ic_payable": "2100"})
must(api("POST", "/accounts", ACCOUNTANT, {"chart_id": "MAIN", "code": "1190", "name": "Due from group", "type": "asset", "subtype": "ic"}))
must(api("POST", "/accounts", ACCOUNTANT, {"chart_id": "MAIN", "code": "2190", "name": "Due to group", "type": "liability", "subtype": "ic"}))
must(api("POST", "/companies", ACCOUNTANT, {"company_code": "DE01", "name": "GmbH", "country": "DE", "functional_currency": "EUR", "chart_id": "MAIN"}))
must(api("POST", "/fiscal-years/open-year", ACCOUNTANT, {"fiscal_year": "FY2026", "company_code": "DE01"}))
must(api("POST", "/fx-rates", ACCOUNTANT, {"from_currency": "GBP", "to_currency": "EUR", "rate": "1.150000", "rate_date": "2026-10-01"}))
for purpose, code, partner in (("ic_receivable", "1190", "DE01"), ("ic_payable", "2190", "UK01"), ("ic_receivable", "1190", "UK01"),
("ic_payable", "2190", "DE01")):
must(api("POST", "/account-map", ACCOUNTANT, {"chart_id": "MAIN", "purpose": purpose, "qualifier": partner,
"account_id": str(account_id(code)), "valid_from": "2026-01-01"}))
must(api("POST", "/companies", ACCOUNTANT, {"company_code": "DE01", "bp_id": "7001"}))
def recharge(who=ACCOUNTANT, amount="1000.00"):
return api("POST", "/journals/propose-cross-company", who, {"journal_date": "2026-10-20", "currency": "GBP", "memo": "software licence",
"lines": [{"company_code": "UK01", "account_id": str(account_id("1200")), "credit": amount},
{"company_code": "DE01", "account_id": str(account_id("6400")), "debit": amount}]})
def test_one_proposal_shows_the_due_to_and_due_from_lines_and_posts_in_both_companies_in_one_approval():
two_companies()
p = must(recharge())
by = {j["company_code"]: j for j in p["journals"]}
assert [(ln["purpose"], ln["debit"], ln["credit"], ln["partner_company"]) for ln in by["UK01"]["lines"] if ln["purpose"]] == \
[("ic_receivable", "1000.00", "0.00", "DE01")]
de = by["DE01"]["lines"]
assert [ln["purpose"] for ln in de] == ["", "ic_payable"] and de[1]["credit"] == "1000.00" and de[1]["functional_credit"] == "1150.00"
assert by["DE01"]["functional_total"] == "1150.00" and by["UK01"]["functional_total"] == "1000.00"
refused(api("POST", f"/journals/{p['journal_ids']['UK01']}/approve-manual", ACCOUNTANT, {}), 409, "[second_person]")
out = must(api("POST", f"/journals/{p['journal_ids']['DE01']}/approve-manual", ACCOUNTANT2, {}))
assert {j["status"] for j in out["journals"]} == {"posted"}
uk = must(api("GET", "/reports/trial-balance", ACCOUNTANT, {"company_code": "UK01", "as_at": "2026-10-31"}))
de = must(api("GET", "/reports/trial-balance", ACCOUNTANT, {"company_code": "DE01", "as_at": "2026-10-31"}))
assert Decimal(uk["difference"]) == 0 and Decimal(de["difference"]) == 0 and de["total_debit"] == "1150.00"
# ids derive from the document and the company
assert p["journal_ids"]["UK01"] == str(seam.derived_id(int(p["ic_document_id"]), "UK01"))
def test_the_second_approver_applies_to_the_whole_document_and_a_rejection_rejects_both():
two_companies()
p = must(recharge())
must(api("POST", f"/journals/{p['journal_ids']['UK01']}/reject-manual", ACCOUNTANT2, {"reason": "wrong entity"}))
for company in ("UK01", "DE01"):
assert must(api("GET", f"/journals/{p['journal_ids'][company]}", ACCOUNTANT))["status"] == "rejected"
def test_a_closed_period_in_either_company_refuses_both_and_the_pair_must_be_mapped():
two_companies()
close_through("2026-09", company="DE01")
out = api("POST", "/journals/propose-cross-company", ACCOUNTANT, {"journal_date": "2026-09-10", "currency": "GBP",
"lines": [{"company_code": "UK01", "account_id": str(account_id("1200")), "credit": "5.00"},
{"company_code": "DE01", "account_id": str(account_id("6400")), "debit": "5.00"}]})
refused(out, 409, "[ic_periods_open]", "DE01")
books_unmapped = api("POST", "/journals/propose-cross-company", ACCOUNTANT, {"journal_date": "2026-10-20", "currency": "GBP",
"lines": [{"company_code": "UK01", "account_id": str(account_id("1200")), "credit": "5.00"},
{"company_code": "FR01", "account_id": str(account_id("6400")), "debit": "5.00"}]})
refused(books_unmapped, 404, "FR01")
def test_a_person_of_one_company_cannot_touch_the_other():
two_companies()
must(api("POST", "/company-access", OWNER, {"principal_sub": "sub-debk", "company_code": "DE01", "valid_from": "2026-01-01"}))
refused(recharge(who=DE_BK), 404, "UK01")
def test_reversing_a_cross_company_journal_reverses_its_partners():
two_companies()
p = must(recharge())
must(api("POST", f"/journals/{p['journal_ids']['DE01']}/approve-manual", ACCOUNTANT2, {}))
must(api("POST", f"/journals/{p['journal_ids']['UK01']}/reverse-journal", ACCOUNTANT, {"reason": "wrong"}))
for c in ("UK01", "DE01"):
tbr = must(api("GET", "/reports/trial-balance", ACCOUNTANT, {"company_code": c, "as_at": "2026-10-31"}))
assert all(Decimal(r["closing"]) == 0 for r in tbr["rows"])
# ── GL-52 ────────────────────────────────────────────────────────────────────────────────────────────────────────────
def neighbours_with(ar_docs, pur_docs):
ar = FastAPI()
pur = FastAPI()
@ar.get("/intercompany/open")
def ar_open(company_code: str, trading_partner: str, as_at: str):
return {"company_code": company_code, "trading_partner": trading_partner, "as_at": as_at, "documents": ar_docs}
@pur.get("/intercompany/open")
def pur_open(company_code: str, trading_partner: str, as_at: str):
return {"company_code": company_code, "trading_partner": trading_partner, "as_at": as_at, "documents": pur_docs}
seam.register_local("ar", ar)
seam.register_local("purchasing", pur)
def test_matching_by_document_then_by_reference_and_a_break_with_its_fix():
two_companies()
p = must(recharge())
must(api("POST", f"/journals/{p['journal_ids']['DE01']}/approve-manual", ACCOUNTANT2, {}))
neighbours_with([{"invoice_id": "i1", "number": "INV-90", "currency": "GBP", "total": "500.00", "outstanding": "500.00"}],
[{"bill_id": "b1", "ic_reference": "INV-90", "currency": "GBP", "total": "498.00", "outstanding": "498.00"}])
m = must(api("POST", "/intercompany/match", ACCOUNTANT, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"}))
assert m["match_id"] == "UK01:DE01:2026-10-31"
assert any(x["key"] == f"ic:{p['ic_document_id']}" for x in m["matched"])
assert any(x["key"] == "balance:GBP" and x["amount"] == "1000.00" for x in m["matched"])
b = m["breaks"][0]
assert b["key"] == "INV-90" and b["difference"] == "2.00" and b["within_tolerance"] is False and "correct the bill" in b["fix"]
# tolerance 1.00: a 2.00 break needs accepting, by a second person
refused(api("POST", "/intercompany/accept-difference", ACCOUNTANT, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31",
"key": "INV-90", "reason": "bill corrected next month"}), 409, "[second_person]")
acc = must(api("POST", "/intercompany/accept-difference", ACCOUNTANT2, {"company_code": "UK01", "partner_company": "DE01",
"as_at": "2026-10-31", "key": "INV-90", "reason": "bill corrected next month"}))
assert acc["breaks"][0]["accepted"] is True and acc["breaks"][0]["reason"]
# re-running keeps what was accepted
again = must(api("POST", "/intercompany/match", ACCOUNTANT, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"}))
assert again["breaks"][0]["accepted"] is True
# consolidation reads it
from conftest import CONSOLIDATION
r = must(api("GET", "/intercompany/reconciliation", CONSOLIDATION, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"}))
assert r["breaks"][0]["accepted"] is True
refused(api("GET", "/intercompany/reconciliation", CONSOLIDATION, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-09-30"}), 404,
"run match_intercompany")
assert api("GET", "/intercompany/reconciliation", ACCOUNTANT, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"})[0] == 403
def test_the_close_gate_ic_matched_refuses_until_the_pair_is_matched_and_the_break_accepted():
two_companies()
p = must(recharge())
must(api("POST", f"/journals/{p['journal_ids']['DE01']}/approve-manual", ACCOUNTANT2, {}))
close_through("2026-09")
g = must(api("GET", "/periods/2026-10/readiness", ACCOUNTANT))
assert next(x for x in g["gates"] if x["gate"] == "ic_matched")["met"] is False
must(api("POST", "/intercompany/match", ACCOUNTANT, {"company_code": "UK01", "partner_company": "DE01", "as_at": "2026-10-31"}))
g = must(api("GET", "/periods/2026-10/readiness", ACCOUNTANT))
assert next(x for x in g["gates"] if x["gate"] == "ic_matched")["met"] is True