dodil/erp-brick-payments

DODIL ERP brick: payments

tests/test_v4_receipts.py

"""Payments v4, slice A - PAY-16 (the early-settlement discount) and PAY-17 (a customer's remittance advice)."""
from decimal import Decimal

from conftest import CASHIER, SYSTEM, allocate, api, bank, must, receipt, refused, set_control, settle
from sqlalchemy import select

from brick_payments.application import use_cases
from brick_payments.infrastructure import tables as t


def table(tbl):
    with use_cases.repository() as repo:
        return [dict(m) for m in repo.conn.execute(select(tbl)).mappings()]


def terms_invoice(nb, iid="INV-2026-0090", total="1200.00", until="2026-11-12"):
    nb.invoice(iid, 7311, total, number=iid, issued_on="2026-11-02", due_date="2026-12-02", discount_pct="2.00", discount_until=until)


# ── PAY-16 ───────────────────────────────────────────────────────────────────────────────────────────

def test_PAY_16_the_difference_is_offered_as_the_discount_and_invoicing_credits_it_at_the_invoices_rates(nb):
    terms_invoice(nb)
    pid = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    got = must(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))
    assert got["discount"] == "24.00" and got["mode"] == "credit_note" and (got["credit"]["net"], got["credit"]["tax"], got["credit"]["total"]) == ("20.00", "4.00", "24.00")
    call = nb.discount_calls[0]
    assert call["invoice_id"] == "INV-2026-0090" and call["body"] == {"payment_id": pid, "received_on": "2026-11-10", "discount_amount": "24.00"}
    # invoicing then allocates its credit through POST /allocations/from-source (PAY-41): the invoice is paid
    nb.credit_note("cn:1", 7311, "24.00")
    must(api("POST", "/allocations/from-source", SYSTEM, {"source_kind": "credit_note", "source_id": "cn:1", "invoice_id": "INV-2026-0090", "amount": "24.00"}))
    assert settle("INV-2026-0090")["INV-2026-0090"]["outstanding"] == "0.00"


def test_PAY_16_a_receipt_after_the_window_gets_no_discount_and_the_difference_stays_outstanding(nb):
    terms_invoice(nb)
    pid = receipt(amount="1176.00", received_on="2026-11-13", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    body = refused(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}), 409,
                   "after the discount ended on 2026-11-12")
    assert "stays outstanding" in str(body) and not nb.discount_calls
    assert settle("INV-2026-0090")["INV-2026-0090"]["outstanding"] == "24.00"


def test_PAY_16_the_window_is_judged_by_the_banks_date_not_by_today(nb):
    terms_invoice(nb)                                           # today is 2026-11-20, long after the 12th - the BANK's date was the 10th
    pid = receipt(amount="1176.00", received_on="2026-11-12", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    assert must(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))["discount"] == "24.00"


def test_PAY_16_the_discount_is_exactly_the_terms_percentage_of_the_gross_outstanding(nb):
    terms_invoice(nb)
    pid = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    refused(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090", "discount_amount": "25.00"}), 409,
            "25.00 is not 2.00% of the gross outstanding 1,200.00")
    short = receipt(amount="1100.00", received_on="2026-11-10", reference="R2")["payment_id"]
    nb.invoice("INV-B", 7311, "1200.00", number="INV-B", discount_pct="2.00", discount_until="2026-11-12")
    must(allocate(short, [("INV-B", "1100.00")]))
    refused(api("POST", f"/receipts/{short}/settlement-discount", CASHIER, {"invoice_id": "INV-B"}), 409, "The receipt paid 1,100.00")
    assert not nb.discount_calls


def test_PAY_16_an_invoice_with_no_terms_has_no_discount(nb):
    nb.invoice("INV-1", 7311, "100.00")
    pid = receipt(amount="98.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-1", "98.00")]))
    refused(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-1"}), 409, "no early-settlement terms")


def test_PAY_16_one_discount_per_invoice_and_asking_again_for_the_same_receipt_is_a_replay(nb):
    terms_invoice(nb)
    a = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(a, [("INV-2026-0090", "1176.00")]))
    must(api("POST", f"/receipts/{a}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))
    assert must(api("POST", f"/receipts/{a}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))["unchanged"] is True
    assert len(nb.discount_calls) == 1
    b = receipt(amount="1176.00", received_on="2026-11-10", reference="R2")["payment_id"]
    refused(api("POST", f"/receipts/{b}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}), 409, "already took its settlement discount")


def test_PAY_16_the_discount_is_claimed_on_the_invoice_in_the_same_transaction(nb):
    terms_invoice(nb)
    a = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(a, [("INV-2026-0090", "1176.00")]))
    must(api("POST", f"/receipts/{a}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))
    assert [(r["claim_key"], r["n"], r["ref"]) for r in table(t.pay_claim) if r["claim_scope"] == "discount"] == [("INV-2026-0090", 1, a)]


def test_PAY_16_invoicings_refusal_travels_in_its_own_words(nb):
    terms_invoice(nb)
    pid = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    nb.discount_refusal = {"gate": "within_discount_window", "reason": "received on 2026-11-10, after the discount ended on 2026-11-09", "fix": "none"}
    refused(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}), 409, "within_discount_window")
    assert [r for r in table(t.pay_claim) if r["claim_scope"] == "discount"] == []              # the refusal rolled the claim back


def test_PAY_16_under_invoice_wording_the_discount_is_an_allocation_tax_reads(nb):
    terms_invoice(nb)
    set_control("settlement_discount_vat", "invoice_wording")
    pid = receipt(amount="1176.00", received_on="2026-11-10", reference="R1")["payment_id"]
    must(allocate(pid, [("INV-2026-0090", "1176.00")]))
    got = must(api("POST", f"/receipts/{pid}/settlement-discount", CASHIER, {"invoice_id": "INV-2026-0090"}))
    assert got["mode"] == "invoice_wording" and not nb.discount_calls                        # no credit note is asked for
    rows = [r for r in table(t.pay_allocation) if r["source_kind"] == "settlement_discount"]
    assert [(r["source_id"], r["invoice_id"], str(r["amount"]), r["seq"]) for r in rows] == [(pid, "INV-2026-0090", "24.00", 1)]
    assert settle("INV-2026-0090")["INV-2026-0090"]["outstanding"] == "0.00"


def test_PAY_16_a_cashier_cannot_switch_the_wording_mode_it_is_a_control(nb):
    from conftest import person
    refused(api("POST", "/settings/settlement_discount_vat", person("c@shop.example", ["pay:settings:write"]), {"value": "invoice_wording"}), 403,
            "pay:settings:controls")


# ── PAY-17 ───────────────────────────────────────────────────────────────────────────────────────────

CSV = ("invoice_ref,amount,deduction_amount,deduction_reason\nINV-1,1000.00,,\nINV-2,600.00,100.00,damaged on arrival\nINV-9,50.00,,\n")


def remit_setup(nb, amount="1550.00"):
    nb.invoice("INV-1", 7311, "1000.00", number="INV-1")
    nb.invoice("INV-2", 7311, "600.00", number="INV-2")
    return receipt(amount=amount, reference="ACME", remittance_text="ACME LTD PAYMENT")["payment_id"]


def post_csv(content=CSV, ref="REM-1", **more):
    return must(api("POST", "/remittances/import", CASHIER, {"format": "csv", "content": content, "remittance_ref": ref, **more}))


def test_PAY_17_the_remittance_is_linked_and_opens_a_suggestion_with_model_remittance(nb):
    pid = remit_setup(nb)
    got = post_csv()
    assert got["linked"] and got["payment_id"] == pid and got["unknown_invoices"] == ["INV-9"]
    s = got["suggestion"]
    assert (s["model"], s["status"], s["proposed_lines"]) == ("remittance", "open", [{"invoice_id": "INV-1", "amount": "1000.00"}, {"invoice_id": "INV-2", "amount": "500.00"}])
    assert s["evidence"]["unknown_invoices"] == ["INV-9"]                                    # an unknown number is listed, never guessed
    assert table(t.pay_allocation) == []                                                     # nothing allocated until a person decides
    lines = {(r["remittance_id"], r["line_no"]): r["status"] for r in table(t.pay_remittance)}
    assert lines == {("REM-1", 1): "imported", ("REM-1", 2): "imported", ("REM-1", 3): "unknown_invoice"}


def test_PAY_17_a_deduction_becomes_a_dispute_with_the_customers_reason_never_a_write_off(nb):
    remit_setup(nb)
    got = post_csv()
    assert got["disputes_opened"] == [{"invoice_id": "INV-2", "amount": "100.00", "reason": "damaged on arrival"}]
    disputes = [r for r in table(t.pay_collection) if r["kind"] == "dispute"]
    assert [(d["invoice_id"], "damaged on arrival" in d["reason"]) for d in disputes] == [("INV-2", True)]
    assert nb.writeoffs == []


def test_PAY_17_accepting_it_as_in_pay_14_allocates_with_decided_how_remittance(nb):
    pid = remit_setup(nb)
    sid = post_csv()["suggestion"]["suggestion_id"]
    got = must(api("POST", f"/receipts/{pid}/decide-match", CASHIER, {"decision": "accept", "suggestion_id": sid}))
    assert got["allocated"] == "1500.00"
    rows = table(t.pay_allocation)
    assert {(r["invoice_id"], str(r["amount"]), r["decided_how"], r["suggestion_id"]) for r in rows} == {("INV-1", "1000.00", "remittance", sid), ("INV-2", "500.00", "remittance", sid)}
    assert settle("INV-2")["INV-2"]["outstanding"] == "100.00"                              # the deduction stays owed, disputed
    assert {r["status"] for r in table(t.pay_remittance)} == {"applied", "unknown_invoice"}


def test_PAY_17_apply_remittance_does_it_in_one_step_and_asking_again_changes_nothing(nb):
    remit_setup(nb)
    post_csv()
    got = must(api("POST", "/remittances/REM-1/apply", CASHIER))
    assert got["status"] == "applied" and got["allocated"] == "1500.00"
    again = must(api("POST", "/remittances/REM-1/apply", CASHIER))
    assert again["unchanged"] is True and len(table(t.pay_allocation)) == 2


def test_PAY_17_a_rejected_remittance_suggestion_is_final(nb):
    pid = remit_setup(nb)
    sid = post_csv()["suggestion"]["suggestion_id"]
    must(api("POST", f"/receipts/{pid}/decide-match", CASHIER, {"decision": "reject", "suggestion_id": sid}))
    refused(api("POST", "/remittances/REM-1/apply", CASHIER), 409, "no open suggestion")
    refused(api("POST", f"/receipts/{pid}/decide-match", CASHIER, {"decision": "accept", "suggestion_id": sid}), 409, "already rejected")
    assert table(t.pay_allocation) == []


def test_PAY_17_importing_the_same_file_again_adds_nothing(nb):
    remit_setup(nb)
    post_csv()
    again = post_csv()
    assert again["imported"] is False and len(table(t.pay_remittance)) == 3 and len(table(t.pay_match_suggestion)) == 1


def test_PAY_17_a_remittance_that_matches_no_single_receipt_is_kept_unlinked_and_can_be_pointed_at_one(nb):
    nb.invoice("INV-1", 7311, "1000.00", number="INV-1")
    got = post_csv("invoice_ref,amount\nINV-1,1000.00\n", ref="REM-2")
    assert got["linked"] is False and "payment_id" in got["note"]
    pid = receipt(amount="900.00", reference="x")["payment_id"]
    refused(api("POST", "/remittances/import", CASHIER, {"format": "csv", "content": "invoice_ref,amount\nINV-1,1000.00\n", "remittance_ref": "REM-3",
                                                         "payment_id": pid}), 409, "not the same payment")


def test_PAY_17_a_camt054_with_structured_remittance_is_read(nb):
    pid = remit_setup(nb)
    xml = ("<?xml version='1.0'?><Document xmlns='urn:iso:std:iso:20022:tech:xsd:camt.054.001.02'><BkToCstmrDbtCdtNtfctn><Ntfctn>"
           "<Ntry><NtryRef>N-77</NtryRef><Amt Ccy='GBP'>1550.00</Amt><CdtDbtInd>CRDT</CdtDbtInd><BookgDt><Dt>2026-11-20</Dt></BookgDt><NtryDtls><TxDtls>"
           "<RltdPties><Dbtr><Nm>Acme Ltd</Nm></Dbtr></RltdPties><RmtInf>"
           "<Strd><RfrdDocInf><Nb>INV-1</Nb></RfrdDocInf><RfrdDocAmt><RmtdAmt Ccy='GBP'>1000.00</RmtdAmt></RfrdDocAmt></Strd>"
           "<Strd><RfrdDocInf><Nb>INV-2</Nb></RfrdDocInf><RfrdDocAmt><RmtdAmt Ccy='GBP'>600.00</RmtdAmt><AdjstmntAmtAndRsn><Amt Ccy='GBP'>100.00</Amt>"
           "<AddtlInf>short delivery</AddtlInf></AdjstmntAmtAndRsn></RfrdDocAmt></Strd>"
           "<Strd><RfrdDocInf><Nb>INV-9</Nb></RfrdDocInf><RfrdDocAmt><RmtdAmt Ccy='GBP'>50.00</RmtdAmt></RfrdDocAmt></Strd>"
           "</RmtInf></TxDtls></NtryDtls></Ntry></Ntfctn></BkToCstmrDbtCdtNtfctn></Document>")
    got = must(api("POST", "/remittances/import", CASHIER, {"format": "camt054", "content": xml}))
    assert got["remittance_id"] == "N-77" and got["payment_id"] == pid and got["disputes_opened"][0]["reason"] == "short delivery"


def test_PAY_17_a_bai2_lockbox_creates_the_receipts_itself_and_the_same_file_again_adds_nothing(nb):
    bank()
    bai = ("01,BANK,DODIL,261120,0900,1,,,2/\n02,DODIL,BANK,1,261119,0900,GBP,/\n03,12345678,GBP/\n"
           "16,142,125000,0,BNK-1,CHQ-501,ACME INV-1/\n88,PAYMENT FOR INV-1/\n16,142,7550,0,BNK-2,CHQ-502,BETA/\n49,132550,3/\n98,132550,1,5/\n99,132550,1,7/\n")
    got = must(api("POST", "/remittances/import", CASHIER, {"format": "bai2", "content": bai, "bank_account_id": "BANK1"}))
    assert got["created"] == 2 and sorted(got["receipt_ids"]) == [f"lockbox:BANK1:{got['batch']}:CHQ-501", f"lockbox:BANK1:{got['batch']}:CHQ-502"]
    r = must(api("GET", f"/receipts/lockbox:BANK1:{got['batch']}:CHQ-501", CASHIER))
    assert (r["amount"], r["method"], r["received_on"]) == ("1250.00", "lockbox", "2026-11-19")
    again = must(api("POST", "/remittances/import", CASHIER, {"format": "bai2", "content": bai, "bank_account_id": "BANK1"}))
    assert again["created"] == 0 and again["skipped_duplicate"] == 2 and len(table(t.pay_receipt)) == 2


def test_PAY_17_only_whoever_may_import_remittances_does_and_a_bad_file_is_a_plain_refusal(nb):
    from conftest import SALES
    refused(api("POST", "/remittances/import", SALES, {"format": "csv", "content": CSV, "remittance_ref": "R"}), 403, "pay:remittance:import")
    refused(api("POST", "/remittances/import", CASHIER, {"format": "csv", "content": "a,b\n1,2\n", "remittance_ref": "R"}), 409, "invoice_ref")
    refused(api("POST", "/remittances/import", CASHIER, {"format": "csv", "content": "invoice_ref,amount,deduction_amount\nINV-1,10.00,5.00\n", "remittance_ref": "R"}),
            409, "deduction with no reason")
    refused(api("POST", "/remittances/import", CASHIER, {"format": "csv", "content": "invoice_ref,amount\nINV-1,1e30\n", "remittance_ref": "R"}), 409, "not an amount")
    assert Decimal("1") and SYSTEM