dodil/erp-brick-payments
DODIL ERP brick: payments
tests/test_v6.py
"""Spec v6: PAY-49 (a cash bag from another module), PAY-50 (expected card settlements), PAY-51 (payouts to a person), the tool names."""
from decimal import Decimal as D
import pytest
from conftest import APPROVER, CASHIER, SYSTEM, api, bank, must, on_day, refused, set_control
from erp_platform import seam
from fastapi import FastAPI
from test_v4_statements import camt, e, imp
BAG = {"batch_ref": "dep:pos:LON-SOHO:004513", "kind": "cash", "bank_account_id": "BANK1", "company_code": "UK01", "currency": "GBP", "amount": "1240.00",
"expected_on": "2026-11-16", "clearing_purpose": "cash_in_transit", "source_module": "pos"}
def journals(nb, event):
return [j for j in nb.journals.values() if j["event"] == event]
def shape(j):
return {(ln.get("purpose") or ln.get("account_id")): (ln.get("debit"), ln.get("credit")) for ln in j["lines"]}
# ── PAY-49 ───────────────────────────────────────────────────────────────────────────────────────
def test_pay49_a_cash_bag_is_one_deposited_batch_matched_to_the_one_credit_and_posts_bank_against_cash_in_transit(nb):
bank("BANK1", gl="1200")
s, a = api("POST", "/deposit-batches/expected", SYSTEM, BAG)
assert s == 201 and a == {"batch_id": "dep:pos:LON-SOHO:004513", "status": "deposited", "total": "1240.00", "replay": False}
s, again = api("POST", "/deposit-batches/expected", SYSTEM, BAG)
assert s == 201 and again["replay"] is True
assert api("POST", "/deposit-batches/expected", SYSTEM, {**BAG, "amount": "999.00"})[0] == 409
s, b = api("POST", "/deposit-batches/expected", SYSTEM, {**BAG, "batch_ref": "dep:pos:LON-SOHO:9", "bank_account_id": "NOPE"})
assert s == 422 and b["gate"] == "bank_account_receives"
got = imp(camt([e("K1", "CRDT", "1240.00", day="2026-11-16", text="CASH PAID IN")], stmt_id="K-1"))
assert got["created"] == 0 and got["matched_to_deposit_batch"] == 1 # the credit is the bag, never a receipt
(j,) = journals(nb, "deposit_match")
assert shape(j) == {"1200": ("1240.00", None), "cash_in_transit": (None, "1240.00")}
assert must(api("GET", "/deposit-batches/dep:pos:LON-SOHO:004513", CASHIER))["lifecycle"] == "matched"
assert api("POST", "/deposit-batches/dep:pos:LON-SOHO:004513/match", CASHIER, {"line_id": "bank:BANK1:K1"})[0] == 409 # the line is already matched
assert journals(nb, "deposit_difference") == []
def test_pay49_a_counted_difference_needs_an_approver_who_is_not_the_cashier_and_posts_cash_over_short(nb):
bank("BANK1", gl="1200")
must_post = lambda body: api("POST", "/deposit-batches/expected", SYSTEM, body) # noqa: E731
assert must_post({**BAG, "batch_ref": "dep:pos:LON-SOHO:004514", "amount": "600.00"})[0] == 201
imp(camt([e("K2", "CRDT", "597.50", day="2026-11-19", text="CASH PAID IN 004514")], stmt_id="K-2"))
body = refused(api("POST", "/deposit-batches/dep:pos:LON-SOHO:004514/match", CASHIER, {"line_id": "bank:BANK1:K2"}), 409, "batch_total_equals_line")
assert body["gate"] == "batch_total_equals_line" and "597.50" in body["reason"]
path = "/deposit-batches/dep:pos:LON-SOHO:004514/accept-difference"
refused(api("POST", path, CASHIER, {"line_id": "bank:BANK1:K2", "reason": "bank counted less"}), 403, "pay:cash_journal:approve")
must(api("POST", path, APPROVER, {"line_id": "bank:BANK1:K2", "reason": "the bank counted 597.50 on a recount"}))
(j,) = journals(nb, "deposit_difference")
assert shape(j) == {"1200": ("597.50", None), "cash_over_short": ("2.50", None), "cash_in_transit": (None, "600.00")}
assert D(j["lines"][0]["debit"]) + D(j["lines"][1]["debit"]) == D(j["lines"][2]["credit"])
# ── PAY-50 ───────────────────────────────────────────────────────────────────────────────────────
def settlement(ref, gross, shop="LON-SOHO", provider="ACQ1"):
return {"settlement_ref": f"card:pos:{shop}:{ref}", "provider": provider, "company_code": "UK01", "currency": "GBP", "gross": gross,
"business_date": "2026-11-14", "transaction_count": 41, "source_module": "pos"}
def providers(value="ACQ1"):
must(api("POST", "/settings/card_providers", APPROVER, {"value": value}))
def test_pay50_one_credit_settles_several_settlements_whole_with_fees_split_pro_rata_and_card_clearing_cleared(nb):
bank("BANK1", gl="1200")
providers()
s, a = api("POST", "/card-settlements/expected", SYSTEM, settlement("T1-2026-11-14", "2310.40"))
assert s == 201 and a == {"settlement_ref": "card:pos:LON-SOHO:T1-2026-11-14", "status": "expected", "replay": False}
assert api("POST", "/card-settlements/expected", SYSTEM, settlement("T1-2026-11-14", "2310.40"))[1]["replay"] is True
assert api("POST", "/card-settlements/expected", SYSTEM, settlement("T1-2026-11-14", "1000.00", shop="LON-KX"))[0] == 201
s, b = api("POST", "/card-settlements/expected", SYSTEM, settlement("X", "10.00", provider="NOPE"))
assert s == 422 and b["gate"] == "provider_known"
imp(camt([e("K3", "CRDT", "3267.69", day="2026-11-17", text="ACQ1 PAYOUT", name="ACQ1")], stmt_id="K-3"))
out = must(api("POST", "/card-settlements/match", CASHIER, {"line_id": "bank:BANK1:K3", "settlement_refs": ["card:pos:LON-SOHO:T1-2026-11-14", "card:pos:LON-KX:T1-2026-11-14"]}))
assert out["state"] == "matched" and out["fees"] == "42.71" and out["fee_pct"] == "1.2902"
(j,) = journals(nb, "card_settlement")
assert shape(j) == {"1200": ("3267.69", None), "card_fees": ("42.71", None), "card_clearing": (None, "3310.40")}
assert [ln["tax_code"] for ln in j["lines"] if ln.get("purpose") == "card_fees"] == ["E"]
rows = {r["settlement_ref"]: r for r in must(api("GET", "/card-settlements", CASHIER, {"lifecycle": "matched"}))["settlements"]}
assert (rows["card:pos:LON-SOHO:T1-2026-11-14"]["fees"], rows["card:pos:LON-KX:T1-2026-11-14"]["fees"]) == ("29.81", "12.90") # pro rata, the rest to the last
assert D("29.81") + D("12.90") == D("42.71")
def test_pay50_fees_above_the_limit_wait_for_an_approver_who_did_not_match_it_and_may_become_a_claim(nb):
bank("BANK1", gl="1200")
providers()
api("POST", "/card-settlements/expected", SYSTEM, settlement("T9", "1000.00"))
imp(camt([e("K4", "CRDT", "950.00", day="2026-11-17", text="ACQ1 PAYOUT", name="ACQ1")], stmt_id="K-4"))
held = must(api("POST", "/card-settlements/match", CASHIER, {"line_id": "bank:BANK1:K4", "settlement_refs": ["card:pos:LON-SOHO:T9"]}))
assert held["state"] == "differs" and held["gate"] == "fees_within_limit" and journals(nb, "card_settlement") == []
assert [r["lifecycle"] for r in must(api("GET", "/card-settlements", CASHIER))["settlements"]] == ["differs"]
refused(api("POST", "/card-settlements/accept-difference", CASHIER, {"line_id": "bank:BANK1:K4", "reason": "ok"}), 403, "pay:cash_journal:approve")
must(api("POST", "/card-settlements/accept-difference", APPROVER, {"line_id": "bank:BANK1:K4", "reason": "the provider's contract rate", "claim": True}))
(j,) = journals(nb, "card_settlement")
assert shape(j) == {"1200": ("950.00", None), "card_claim": ("50.00", None), "card_clearing": (None, "1000.00")}
set_control("card_fee_max_pct", "6.00") # the limit is the owner's control
api("POST", "/card-settlements/expected", SYSTEM, settlement("T10", "1000.00"))
imp(camt([e("K5", "CRDT", "950.00", day="2026-11-18", text="ACQ1 PAYOUT", name="ACQ1")], stmt_id="K-5"))
assert must(api("POST", "/card-settlements/match", CASHIER, {"line_id": "bank:BANK1:K5", "settlement_refs": ["card:pos:LON-SOHO:T10"]}))["state"] == "matched"
# ── PAY-51 ───────────────────────────────────────────────────────────────────────────────────────
class World:
"""hr (the payee's bank account, payments' client only) and expenses (the 'paid' report), answering their signed routes."""
def __init__(self):
self.told, self.hr_reads = [], []
self.expenses_down = False
hr, exp = FastAPI(), FastAPI()
@hr.get("/persons/{person_id}/bank-account")
def account(person_id: str, on: str = ""):
self.hr_reads.append((person_id, on))
if person_id == "P-NONE":
from fastapi.responses import JSONResponse
return JSONResponse({"detail": "no bank account"}, status_code=404)
return {"person_id": person_id, "account_holder": "Jane Doe", "sort_code": "20-00-00", "account_number": "55779911", "iban": None, "bic": None,
"valid_from": "2026-01-01"}
@exp.post("/reports/{report_id}/paid")
def paid(report_id: str, body: dict):
if self.expenses_down:
from fastapi.responses import JSONResponse
return JSONResponse({"detail": "down"}, status_code=503)
self.told.append((report_id, body))
return {"report_id": report_id, "status": "paid"}
seam.register_local("hr", hr)
seam.register_local("expenses", exp)
def close(self):
seam.unregister_local("hr")
seam.unregister_local("expenses")
@pytest.fixture
def w():
x = World()
yield x
x.close()
PAYOUT = {"payout_ref": "exp:UK01-EXP-2026-00042:r1:pay", "payee_bp_id": 9001, "person_id": "P-77", "company_code": "UK01", "currency": "GBP", "amount": "352.40",
"due_on": "2026-11-20", "reference": "EXP-2026-00042", "source_module": "expenses", "source_document": "UK01-EXP-2026-00042",
"clearing_purpose": "employee_payable"}
def test_pay51_a_payout_is_requested_approved_exported_with_the_payees_account_from_hr_and_confirmed_from_the_statement(nb, w):
bank("BANK1", gl="1200", purposes="receive,pay")
s, a = api("POST", "/payouts", SYSTEM, PAYOUT)
assert s == 201 and a["status"] == "requested" and a["replay"] is False
assert api("POST", "/payouts", SYSTEM, PAYOUT)[1]["replay"] is True
assert api("POST", "/payouts", SYSTEM, {**PAYOUT, "amount": "1.00"})[0] == 409
s, b = api("POST", "/payouts", SYSTEM, {**PAYOUT, "payout_ref": "exp:X:r1:pay", "currency": "EUR"})
assert s == 422 and b["gate"] == "paying_account"
refused(api("POST", "/payout-runs/export", CASHIER, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}), 409, "No approved payout")
must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/approve", APPROVER, {}))
run = must(api("POST", "/payout-runs/export", CASHIER, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}))
assert run["count"] == 1 and run["total"] == "352.40" and run["format"] == "pain.001" and w.hr_reads == [("P-77", "2026-11-20")]
with __import__("brick_payments.application.use_cases", fromlist=["x"]).repository() as repo:
body = repo.objects.get(run["file_key"])
assert "55779911" in body and "Jane Doe" in body and "352.40" in body # the payee's account is in the FILE ...
shown = str(must(api("GET", "/payouts", CASHIER)))
assert "55779911" not in shown and "20-00-00" not in shown # ... and nowhere else
imp(camt([e("K6", "DBIT", "352.40", day="2026-11-20", text="EXP-2026-00042", name="Jane Doe")], stmt_id="K-6"))
done = must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/confirm-paid", CASHIER, {"line_id": "bank:BANK1:K6"}))
assert done["status"] == "paid" and done["told_source"] is True
(j,) = journals(nb, "payout")
assert shape(j) == {"employee_payable": ("352.40", None), "1200": (None, "352.40")}
((report, told),) = w.told
assert report == "UK01-EXP-2026-00042" and told["payout_ref"] == PAYOUT["payout_ref"] and told["amount"] == "352.40" and told["paid_on"] == "2026-11-20"
assert must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/confirm-paid", CASHIER, {"line_id": "bank:BANK1:K6"}))["status"] == "paid" # a repeat changes nothing
assert len(w.told) == 1
def test_pay51_the_source_is_told_until_it_answers_and_a_returned_payout_reverses_the_journal(nb, w):
bank("BANK1", gl="1200", purposes="receive,pay")
api("POST", "/payouts", SYSTEM, PAYOUT)
must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/approve", APPROVER, {}))
must(api("POST", "/payout-runs/export", CASHIER, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}))
w.expenses_down = True
done = must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/confirm-paid", CASHIER, {"bank_ref": "FPS-1", "paid_on": "2026-11-20"}))
assert done["status"] == "paid" and done["told_source"] is False and w.told == []
w.expenses_down = False
assert must(api("POST", "/payouts/notify", SYSTEM, {})) == {"tried": 1, "told": 1, "still_waiting": 0}
assert len(w.told) == 1
ret = must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/return", CASHIER, {"bank_ref": "RTN-9"}))
assert ret["status"] == "returned"
(j,) = journals(nb, "payout_return")
assert shape(j) == {"1200": ("352.40", None), "employee_payable": (None, "352.40")}
def test_pay51_the_exporter_must_not_be_an_approver_and_hr_must_hold_an_account(nb, w):
from conftest import person
bank("BANK1", gl="1200", purposes="receive,pay")
both = person("both@shop.example", ["pay:payout:approve", "pay:payout:export"])
api("POST", "/payouts", SYSTEM, PAYOUT)
must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/approve", both, {}))
body = refused(api("POST", "/payout-runs/export", both, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}), 409, "exporter_not_approver")
assert body["gate"] == "exporter_not_approver" and w.hr_reads == []
api("POST", "/payouts", SYSTEM, {**PAYOUT, "payout_ref": "exp:NONE:r1:pay", "person_id": "P-NONE"})
must(api("POST", "/payouts/exp:NONE:r1:pay/approve", APPROVER, {}))
body = refused(api("POST", "/payout-runs/export", CASHIER, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}), 409, "payee_bank_known")
assert body["gate"] == "payee_bank_known" # the whole run waits: nothing was exported
assert {p["status"] for p in must(api("GET", "/payouts", CASHIER))["payouts"]} == {"approved"}
def test_pay51_a_payout_is_cancelled_by_the_module_that_asked_or_by_an_approver_and_never_once_exported(nb, w):
bank("BANK1", gl="1200", purposes="receive,pay")
api("POST", "/payouts", SYSTEM, PAYOUT)
api("POST", "/payouts", SYSTEM, {**PAYOUT, "payout_ref": "exp:B:r1:pay"})
assert must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/cancel", SYSTEM, {"reason": "claim reversed"})) == {"payout_ref": PAYOUT["payout_ref"], "status": "cancelled"}
assert must(api("POST", f"/payouts/{PAYOUT['payout_ref']}/cancel", SYSTEM, {"reason": "again"}))["status"] == "cancelled"
assert must(api("POST", "/payouts/exp:B:r1:pay/withdraw", APPROVER, {"reason": "paid another way"}))["status"] == "cancelled"
api("POST", "/payouts", SYSTEM, {**PAYOUT, "payout_ref": "exp:C:r1:pay"})
must(api("POST", "/payouts/exp:C:r1:pay/approve", APPROVER, {}))
must(api("POST", "/payout-runs/export", CASHIER, {"bank_account_id": "BANK1", "pay_date": "2026-11-20"}))
body = refused(api("POST", "/payouts/exp:C:r1:pay/cancel", SYSTEM, {"reason": "late"}), 409, "not_exported")
assert body["gate"] == "not_exported"
def test_the_shared_family_tools_name_the_module_noun():
from brick_payments.application import use_cases
names = {u.name for u in use_cases.PAY.use_cases}
assert {"load_payments_from_odoo", "sign_payments_migration"} <= names and not {"run_odoo_migration", "sign_migration"} & names