dodil/erp-brick-ar
DODIL ERP brick: accounts receivable
tests/test_invariants.py
"""The spec's tests.invariants that the stories do not already prove, each by breaking the rule and watching it refused."""
from decimal import Decimal
from conftest import (
APPROVER,
BILLER,
BP,
SALES,
SYSTEM,
account_map,
api,
approved_credit,
billing,
control,
hand_draft,
issue,
issued,
line,
must,
on_day,
)
from sqlalchemy.exc import IntegrityError
from brick_ar.infrastructure.tables import NumberRepository
def test_a_void_invoice_refuses_allocations(world):
inv = issued(world)
cn = approved_credit(world, issued(world, lines=[line(price="10.00")])["invoice_id"], full=True)
must(api("POST", f"/invoices/{inv['invoice_id']}/void", APPROVER, {"reason": "raised in error"}))
s, b = api("POST", f"/credit-notes/{cn['credit_id']}/allocate", BILLER, {"invoice_id": inv["invoice_id"], "amount": "1.00"})
assert s == 409 and "issued invoice" in b["detail"]
s, b = api("POST", "/writeoffs", BILLER, {"invoice_id": inv["invoice_id"], "amount": "1.00", "reason": "x"})
assert s == 409 and "void" in b["detail"]
s, b = api("POST", "/credit-notes", BILLER, {"invoice_id": inv["invoice_id"], "reason": "x", "full": True})
assert s == 409 and "void" in b["detail"]
def test_amounts_are_never_summed_across_currencies(world):
with on_day("2026-09-11"):
billing()
a = issue(hand_draft()["invoice_id"], world=world)
eur = must(api("POST", "/invoices", BILLER, {"bp_id": BP, "currency": "EUR", "fx_rate": "0.8600", "lines": [line(price="50.00")]}))
issue(eur["invoice_id"], world=world)
out = must(api("GET", f"/customers/{BP}/status", SALES))
assert out["overdue_by_currency"] == {"GBP": "240.00", "EUR": "120.00"} and out["overdue_amount"] is None
s, b = api("POST", "/credit-notes", BILLER, {"invoice_id": a["invoice_id"], "reason": "r", "full": True})
cn = must((s, b))
world.payments.credit_faces[cn["credit_id"]] = Decimal(cn["total"])
must(api("POST", f"/credit-notes/{cn['credit_id']}/approve", APPROVER, {}))
s, b = api("POST", f"/credit-notes/{cn['credit_id']}/allocate", BILLER, {"invoice_id": eur["invoice_id"], "amount": "1.00"})
assert s == 409 and "same customer and the same currency" in b["detail"]
def test_the_functional_currency_total_is_carried_separately(world):
billing()
d = must(api("POST", "/invoices", BILLER, {"bp_id": BP, "currency": "EUR", "fx_rate": "0.8600", "lines": [line(qty="1", price="1000.00")]}))
assert (d["total"], d["total_functional"], d["fx_rate"]) == ("1200.00", "1032.00", "0.8600")
def test_today_is_the_date_in_the_business_time_zone_for_issue_and_every_dated_row(world):
"""At 15:00 UTC on 20 Nov it is already 21 Nov in Auckland: the issue date, and the 'after today' rule, follow the owner's zone."""
with on_day("2026-11-20", hour=15):
billing()
must(control("business_timezone", "Pacific/Auckland"))
inv = issue(hand_draft()["invoice_id"], world=world)
assert inv["issue_date"] == "2026-11-21" and inv["due_date"] == "2026-12-21"
s, b = api("POST", f"/customers/{BP}/billing", APPROVER, {"bill_to_name": "A", "bill_to_address": "x", "invoice_email": "e@x.example",
"terms_id": "NET30", "currency": "GBP", "valid_from": "2026-11-21"})
assert s == 409 and "tomorrow at the earliest" in b["detail"]
cn = approved_credit(world, inv["invoice_id"], lines=[line("c", "1", "10.00")])
assert cn["credit_date"] == "2026-11-21"
def test_a_ledger_mode_switch_starts_on_the_first_of_a_later_month(world):
must(control("ledger_mode", "export", "2026-12-01"))
s, b = control("ledger_mode", "post", "2027-01-15")
assert s == 409 and "first day of a month" in b["detail"]
assert control("ledger_mode", "post", "2027-02-01")[0] == 200
def test_a_lost_race_for_a_number_is_retried_and_numbers_stay_unique(world, monkeypatch):
real, calls = NumberRepository.claim, []
def flaky(self, series_id, document_id, at, **kw):
calls.append(document_id)
if len(calls) == 1:
raise IntegrityError("INSERT INTO ar_number_claims", {}, Exception("UNIQUE constraint failed: ar_number_claims.series_id, ar_number_claims.seq"))
return real(self, series_id, document_id, at, **kw)
monkeypatch.setattr(NumberRepository, "claim", flaky)
billing()
a = issue(hand_draft()["invoice_id"], world=world)
monkeypatch.setattr(NumberRepository, "claim", real)
b = issue(hand_draft()["invoice_id"], world=world)
assert (a["invoice_no"], b["invoice_no"]) == ("INV-2026-0001", "INV-2026-0002") and len(calls) == 2
def test_a_number_format_change_starts_a_new_series_from_its_date(world):
billing()
a = issue(hand_draft()["invoice_id"], world=world)
must(control("invoice_number_format", "BILL/{yyyy}/{seq:5}"))
with on_day("2026-11-21"):
b = issue(hand_draft()["invoice_id"], world=world)
assert (a["invoice_no"], b["invoice_no"]) == ("INV-2026-0001", "BILL/2026/00001")
def test_numbers_restart_each_year_unless_the_owner_says_otherwise(world):
with on_day("2026-12-31"):
billing()
a = issue(hand_draft()["invoice_id"], world=world)
with on_day("2027-01-02"):
b = issue(hand_draft()["invoice_id"], world=world)
assert (a["invoice_no"], b["invoice_no"]) == ("INV-2026-0001", "INV-2027-0001")
def test_the_same_source_yields_one_invoice_even_when_the_payload_differs(world):
billing()
a = must(api("POST", "/invoices/raise", SYSTEM, {"source_kind": "contract", "source_id": "ct:C-9", "bp_id": BP, "currency": "GBP", "lines": [line()]}))
b = must(api("POST", "/invoices/raise", SYSTEM, {"source_kind": "contract", "source_id": "ct:C-9", "bp_id": BP, "currency": "GBP",
"lines": [line(price="999.00")]}))
assert a["invoice_id"] == b["invoice_id"] and b["total"] == a["total"] == "240.00"
def test_an_unconfigured_account_map_is_the_only_thing_that_stops_posting_and_a_later_row_never_rewrites_a_past_posting(world):
account_map()
inv = issued(world)
must(api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {}))
first = next(iter(world.gl.journals.values()))
must(api("POST", "/account-map", APPROVER, {"purpose": "revenue", "account_id": 4100})) # from tomorrow
s, b = api("POST", "/account-map", APPROVER, {"purpose": "revenue", "account_id": 4200, "valid_from": "2026-11-20"})
assert s == 409 and "tomorrow at the earliest" in b["detail"]
must(api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {}))
assert len(world.gl.journals) == 1 and next(iter(world.gl.journals.values())) == first
with on_day("2026-11-21"):
inv2 = issue(hand_draft()["invoice_id"], world=world)
must(api("POST", f"/documents/{inv2['invoice_id']}/post", SYSTEM, {}))
assert {x["account_id"] for j in world.gl.journals.values() for x in j["lines"]} >= {"4000", "4100"}
def test_the_export_and_posting_are_exclusive_by_mode(world):
inv = issued(world)
out = must(api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {}))["results"][0]
assert out["posted"] is False and out["gate"]["gate"] == "account_map"
world.remove("gl")
s, b = api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {})
assert s == 200 and "not post" in b["results"][0]["reason"]
def test_post_pending_posts_every_document_the_ledger_has_not_got(world):
account_map()
a = issued(world)
b = issue(hand_draft()["invoice_id"], world=world)
approved_credit(world, a["invoice_id"], lines=[line("c", "1", "10.00")])
out = must(api("POST", "/documents/post-pending", SYSTEM, {}))
assert out["posted"] == 3 and len(world.gl.journals) == 3
assert must(api("POST", "/documents/post-pending", SYSTEM, {}))["posted"] == 0 and len(world.gl.journals) == 3
assert b["invoice_id"]
def test_post_pending_limit_counts_only_what_is_still_to_post(world):
"""Past `limit` posted documents a new one must still post: the limit was taken before skipping the posted ones."""
account_map()
issued(world), issue(hand_draft()["invoice_id"], world=world)
assert must(api("POST", "/documents/post-pending", SYSTEM, {"limit": 2}))["posted"] == 2
issue(hand_draft()["invoice_id"], world=world)
out = must(api("POST", "/documents/post-pending", SYSTEM, {"limit": 2}))
assert out["posted"] == 1 and len(world.gl.journals) == 3
def test_posting_after_a_crash_between_the_ledger_and_our_record_still_changes_the_ledger_once(world, engine):
"""The ledger answered but our posting record was lost: posting again sends the SAME derived journal id, so the ledger holds one."""
from sqlalchemy import delete
from brick_ar.infrastructure.tables import ar_postings
account_map()
inv = issued(world)
must(api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {}))
with engine.begin() as c:
c.execute(delete(ar_postings))
must(api("POST", f"/documents/{inv['invoice_id']}/post", SYSTEM, {}))
assert len(world.gl.journals) == 1