dodil/erp-brick-payments

DODIL ERP brick: payments

tests/test_domain.py

"""The rules without a database: settlement arithmetic, aging, the matcher, the bank files, the credit hold, reminders."""
from datetime import date, datetime
from decimal import Decimal

import pytest

from brick_payments.domain import bankfile, credit, matching
from brick_payments.domain import collection as coll
from brick_payments.domain import settlement as stl
from brick_payments.domain.errors import RuleBroken
from brick_payments.domain.receipt import Receipt
from brick_payments.domain.refund import Refund
from brick_payments.domain.settlement import Allocation

D = Decimal
TODAY = date(2026, 11, 20)


def alloc(kind, sid, inv, seq, amount, on=TODAY, **kw):
    return Allocation(kind, sid, inv, seq, D(amount), "GBP", on, **kw)


def test_outstanding_is_the_total_less_the_allocations_dated_on_or_before_the_day():
    rows = [alloc("receipt", "R", "I", 1, "400.00", date(2026, 11, 1)), alloc("credit_note", "C", "I", 1, "100.00", date(2026, 11, 10)),
            alloc("receipt", "R", "I", 2, "-400.00", date(2026, 11, 15), reverses_seq=1)]
    assert stl.outstanding(D("1000.00"), rows) == D("1000.00") - D("100.00")
    assert stl.outstanding(D("1000.00"), rows, date(2026, 11, 5)) == D("600.00")
    assert stl.outstanding(D("1000.00"), rows, date(2026, 10, 31)) == D("1000.00")


def test_state_follows_from_the_arithmetic():
    assert stl.state_of(D("100"), D("100"), []) == "open"
    assert stl.state_of(D("100"), D("40"), [alloc("receipt", "R", "I", 1, "60")]) == "part_paid"
    assert stl.state_of(D("100"), D("0"), [alloc("receipt", "R", "I", 1, "100")]) == "paid"
    assert stl.state_of(D("100"), D("0"), [alloc("writeoff", "W", "I", 1, "100")]) == "written_off"


def test_a_receipts_remainder_counts_allocations_reversals_and_refunds_out_of_it():
    rows = [alloc("receipt", "R", "I1", 1, "700.00"), alloc("receipt", "R", "I1", 2, "-200.00", reverses_seq=1),
            alloc("refund", "F", "receipt:R", 1, "100.00"), alloc("receipt", "OTHER", "I1", 1, "999.00")]
    assert stl.source_remainder(D("1000.00"), "receipt", "R", rows) == D("400.00")
    cross = [alloc("receipt", "R", "I1", 1, "850.00", source_amount=D("1000.00"))]
    assert stl.source_remainder(D("1000.00"), "receipt", "R", cross) == D("0.00")          # drawn in ITS currency


def test_aging_buckets_parse_and_label_any_configuration():
    assert stl.parse_buckets("0-30,31-60,61-90,90+") == [(0, 30), (31, 60), (61, 90), (90, None)]
    assert [stl.bucket_label(d, "0-30,31-60,61-90,90+") for d in (1, 30, 31, 60, 61, 90, 91, 400)] == ["0-30", "0-30", "31-60", "31-60", "61-90", "61-90", "90+", "90+"]
    assert stl.bucket_label(5, "0-7,8+") == "0-7"


def test_a_receipt_is_dated_today_or_earlier_and_a_return_not_before_it_arrived():
    r = Receipt.record(payment_id="p", bp_id=1, currency="gbp", amount="10", received_on=date(2026, 11, 18), today=TODAY,
                       recorded_on=date(2026, 11, 18))
    assert (r.currency, r.amount) == ("GBP", D("10.00")) and r.events
    with pytest.raises(RuleBroken, match="after today"):
        Receipt.record(payment_id="p", bp_id=1, currency="GBP", amount="10", received_on=date(2026, 11, 21), today=TODAY)
    with pytest.raises(RuleBroken, match="more than zero"):
        Receipt.record(payment_id="p", bp_id=1, currency="GBP", amount="0", received_on=TODAY, today=TODAY)
    with pytest.raises(RuleBroken, match="before the receipt"):
        r.reverse(returned_on=date(2026, 11, 17), bank_return_ref="x", today=TODAY)
    r.reverse(returned_on=date(2026, 11, 19), bank_return_ref="RTN", today=TODAY)
    assert (r.status, r.reversed_on, r.live) == ("reversed", TODAY, False)
    assert r.live_at(date(2026, 11, 19)) and not r.live_at(TODAY)               # judged by the day WE recorded the return
    late = Receipt.record(payment_id="q", bp_id=1, currency="GBP", amount="10", received_on=date(2026, 11, 18), today=TODAY)
    assert not late.live_at(date(2026, 11, 19)) and late.live_at(TODAY)         # ... and the day WE recorded the receipt (audit 1)


def test_a_refund_walks_forward_only_and_never_approves_itself():
    f = Refund.request(refund_id="f", bp_id=1, currency="GBP", amount="300", credit=D("300"), by="a@x", at=datetime(2026, 11, 20), reason="")
    with pytest.raises(RuleBroken, match="You requested this refund"):
        f.approve(by="A@x", credit=D("300"), at=datetime(2026, 11, 20), on=TODAY)
    with pytest.raises(RuleBroken, match="is requested"):
        f.pay(paid_on=TODAY, bank_ref="b", today=TODAY)
    f.approve(by="b@x", credit=D("300"), at=datetime(2026, 11, 20), on=TODAY)
    with pytest.raises(RuleBroken, match="unallocated credit"):
        Refund.request(refund_id="g", bp_id=1, currency="GBP", amount="300.01", credit=D("300"), by="a@x", at=datetime(2026, 11, 20))
    f.pay(paid_on=TODAY, bank_ref="b", today=TODAY)
    assert f.status == "paid" and [type(e).__name__ for e in f.events] == ["RefundRequested", "RefundApproved", "RefundPaid"]


def test_the_matcher_names_invoices_by_number_or_trailing_digits_and_knows_exact_from_partial():
    inv = [matching.Open("a", "INV-2026-0041", D("4000.00"), date(2026, 11, 1)), matching.Open("b", "INV-2026-0042", D("3500.00"), date(2026, 11, 2)),
           matching.Open("c", "INV-2026-0043", D("3000.00"), date(2026, 11, 3))]
    exact = matching.suggest(amount=D("7500"), remainder=D("7500"), reference="INV-2026-0041 + INV-2026-0042", open_invoices=inv)
    assert [ln["invoice_id"] for ln in exact.lines] == ["a", "b"] and exact.exact and exact.confidence == D("0.97")
    short_form = matching.suggest(amount=D("7500"), remainder=D("7500"), reference="INV41 + INV42", open_invoices=inv)
    assert [ln["invoice_id"] for ln in short_form.lines] == ["a", "b"] and not short_form.exact          # the whole number makes it exact (audit 1)
    wide = matching.suggest(amount=D("10000"), remainder=D("10000"), reference="INV-2026-0041 0042", open_invoices=inv)
    assert [ln["invoice_id"] for ln in wide.lines] == ["a", "b"] and not wide.exact and wide.confidence == D("0.85")
    short = matching.suggest(amount=D("5000"), remainder=D("5000"), reference="INV41 INV42", open_invoices=inv)
    assert [(ln["invoice_id"], ln["amount"]) for ln in short.lines] == [("a", D("4000.00")), ("b", D("1000.00"))] and not short.exact
    assert matching.suggest(amount=D("3000"), remainder=D("3000"), reference="thanks", open_invoices=inv).lines[0]["invoice_id"] == "c"   # amount equals one invoice
    assert matching.suggest(amount=D("1"), remainder=D("1"), reference="the year 2026", open_invoices=inv) is None                     # 2026 names nothing
    assert matching.suggest(amount=D("5"), remainder=D("0"), reference="INV41", open_invoices=inv) is None


def test_camt053_and_csv_parse_credits_and_debits():
    xml = ("<Document xmlns='urn:iso:std:iso:20022:tech:xsd:camt.053.001.02'><BkToCstmrStmt><Stmt>"
           "<Ntry><NtryRef>R1</NtryRef><Amt Ccy='GBP'>12.50</Amt><CdtDbtInd>CRDT</CdtDbtInd><BookgDt><Dt>2026-11-18</Dt></BookgDt></Ntry>"
           "<Ntry><AcctSvcrRef>R2</AcctSvcrRef><Amt Ccy='EUR'>3.00</Amt><CdtDbtInd>DBIT</CdtDbtInd><BookgDt><DtTm>2026-11-19T10:00:00</DtTm></BookgDt></Ntry>"
           "</Stmt></BkToCstmrStmt></Document>")
    lines = bankfile.parse_camt053(xml)
    assert [(ln.entry_ref, ln.amount, ln.currency, ln.credit, ln.booked_on) for ln in lines] == [
        ("R1", D("12.50"), "GBP", True, date(2026, 11, 18)), ("R2", D("3.00"), "EUR", False, date(2026, 11, 19))]
    with pytest.raises(RuleBroken, match="entry reference"):
        bankfile.parse_camt053(xml.replace("<NtryRef>R1</NtryRef>", ""))
    with pytest.raises(RuleBroken, match="not a readable"):
        bankfile.parse_camt053("<nope")
    csv = bankfile.parse_csv("Date,Amount,Ccy,Reference\n2026-11-18,\"1,250.00\",GBP,A\n2026-11-18,-5,GBP,B\n")
    assert [(c.entry_ref, c.amount, c.credit) for c in csv] == [("A", D("1250.00"), True), ("B", D("5"), False)]
    with pytest.raises(RuleBroken, match="needs a column"):
        bankfile.parse_csv("a,b\n1,2\n")


def test_the_hold_is_derived_and_the_people_override_it():
    now = datetime(2026, 11, 20, 9)
    held = credit.derive(today=TODAY, now=now, limit=None, balance_in_limit_currency=D("0"), oldest_overdue_days=61, days_rule=60,
                         over_limit_rule=True, overrides=[])
    assert held.held and held.source == "rule"
    rel = credit.Override("release", now, date(2026, 12, 1), "paid", "owner")
    assert not credit.derive(today=TODAY, now=now, limit=None, balance_in_limit_currency=D("0"), oldest_overdue_days=61, days_rule=60,
                             over_limit_rule=True, overrides=[rel]).held
    manual = credit.Override("hold", now, date(2026, 12, 1), "fraud", "owner")
    assert credit.derive(today=TODAY, now=now, limit=None, balance_in_limit_currency=D("0"), oldest_overdue_days=0, days_rule=60,
                         over_limit_rule=True, overrides=[manual]).source == "manual"
    limits = [credit.Limit(date(2026, 1, 1), D("100"), "GBP"), credit.Limit(date(2026, 6, 1), D("200"), "GBP")]
    assert credit.limit_on(limits, date(2026, 5, 31)).amount == D("100") and credit.limit_on(limits, date(2026, 6, 1)).amount == D("200")
    assert credit.limit_on(limits, date(2025, 12, 31)) is None


def test_reminder_levels_promises_and_disputes():
    levels = [coll.Level(1, 7, mode="suggest"), coll.Level(2, 14), coll.Level(3, 30), coll.Level(4, 60, mode="off")]
    due = date(2026, 11, 1)
    assert coll.next_level(levels, [], "I", due, date(2026, 11, 7)) is None
    assert coll.next_level(levels, [], "I", due, date(2026, 11, 8)).level == 1
    assert coll.next_level(levels, [], "I", due, date(2026, 12, 20)).level == 3                # the highest due; level 4 is off
    seen = [coll.Event("reminder", 3, "I", "I", TODAY, None, "sent")]
    assert coll.next_level(levels, seen, "I", due, date(2026, 12, 20)) is None
    promise = coll.Event("promise", 1, "x", "", TODAY, date(2026, 12, 5))
    assert coll.suppression([promise], "I", date(2026, 12, 5)) and not coll.suppression([promise], "I", date(2026, 12, 6))
    dispute = coll.Event("dispute", 1, "I", "I", TODAY, reason="damaged")
    assert coll.suppression([dispute], "I", TODAY).startswith("an open dispute")
    assert not coll.suppression([dispute, coll.Event("resolution", 1, "I", "I")], "I", TODAY)
    assert coll.state_of(outstanding=D("5"), due_date=due, today=TODAY, events=[dispute], invoice_id="I") == "disputed"
    assert coll.state_of(outstanding=D("0"), due_date=due, today=TODAY, events=[], invoice_id="I") == "resolved"
    assert coll.state_of(outstanding=D("5"), due_date=date(2026, 12, 1), today=TODAY, events=[], invoice_id="I") == "not_due"