dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_stories_downpayments.py
"""Down payments — PUR-47 (request, approve), PUR-48 (paid in a run), PUR-49 (cleared against the final bill)."""
from conftest import (
ALICE,
BUYER,
CLERK,
PAYAPPROVER,
PAYAPPROVER2,
SYSTEM,
api,
bill,
confirmed,
must,
on_day,
receive,
supplier_price,
world,
)
from test_stories_pay import propose
DUE = "2026-11-25"
def order_with_dp(percent="50"):
world()
with on_day("2026-01-01T09:00:00+00:00"):
must(api("POST", "/account-map", ALICE, {"purpose": "supplier_down_payment", "category": "*", "account_id": "1600", "valid_from": "2026-01-02"}))
supplier_price()
o = confirmed(lines=[{"product_id": "COLA-075", "quantity": "10"}])
d = must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/down-payments", BUYER, {"percent": percent, "due_date": DUE, "tax_code": "STD", "tax_amount": "18.00"}))
return o, d
def test_pur47_a_request_is_numbered_a_noted_item_and_bounded_by_the_order(neighbours):
o, d = order_with_dp()
assert d["dp_no"] == "DP-2026-0001" and str(d["net_amount"]) == "90.00" and str(d["gross_amount"]) == "108.00" and d["status"] == "requested"
assert neighbours.gl.journals == [] or not neighbours.gl.posted(d["down_payment_id"], "pay") # nothing posts until paid
s, b = api("POST", f"/purchase-orders/{o['purchase_order_id']}/down-payments", BUYER, {"percent": "60", "due_date": DUE})
assert s == 409 and "90.00 of 180.00 remains" in str(b)
assert api("POST", f"/purchase-orders/{o['purchase_order_id']}/down-payments", CLERK, {"percent": "10", "due_date": DUE})[0] == 403
def test_pur47_the_buyer_cannot_approve_their_own_request_and_a_cancelled_order_takes_none():
world()
supplier_price()
o = confirmed(who=ALICE, lines=[{"product_id": "COLA-075", "quantity": "10"}])
d = must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/down-payments", ALICE, {"percent": "30", "due_date": DUE}))
s, b = api("POST", f"/down-payments/{d['down_payment_id']}/approve", ALICE, {})
assert s == 409 and "second_person" in str(b)
ok = must(api("POST", f"/down-payments/{d['down_payment_id']}/approve", PAYAPPROVER, {}))
assert ok["status"] == "approved" and ok["approved_by_email"] == "finance@shop.example"
must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/cancel", ALICE, {"reason": "no longer needed"}))
assert api("POST", f"/purchase-orders/{o['purchase_order_id']}/down-payments", ALICE, {"percent": "10", "due_date": DUE})[0] == 409
def paid_dp(neighbours):
o, d = order_with_dp()
must(api("POST", f"/down-payments/{d['down_payment_id']}/approve", PAYAPPROVER, {}))
run = must(propose())
must(api("POST", f"/payment-runs/{run['run_id']}/approve", PAYAPPROVER2, {}))
must(api("POST", f"/payment-runs/{run['run_id']}/export", SYSTEM, {}))
pid = run["payments"][0]["payment_id"]
must(api("POST", f"/payments/{pid}/confirm", CLERK, {"bank_reference": "B1", "paid_on": "2026-11-20"}))
return o, d, pid
def test_pur48_an_approved_down_payment_is_paid_in_the_run_and_posts_to_supplier_down_payments(neighbours):
o, d, pid = paid_dp(neighbours)
got = must(api("GET", f"/down-payments/{d['down_payment_id']}", BUYER))
assert got["status"] == "paid" and got["paid_in_advance"] is True and str(got["remaining"]) == "108.00" and "refund" in got["fix"]
(j,) = neighbours.gl.posted(pid, "pay")
assert sorted((x["account_id"], x.get("debit") or x.get("credit")) for x in j["lines"]) == [("1200", "108.00"), ("1600", "108.00")]
def test_pur49_clearing_lowers_the_bill_and_a_bill_waits_for_it_before_a_run(neighbours):
o, d, pid = paid_dp(neighbours)
receive(o, quantity="10")
s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
run = propose()
assert run[0] == 409 or "uncleared down payment" in str(run[1]) # held out of a run until cleared
assert api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "109.00"})[0] == 409
c = must(api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "54.00"}))
assert str(c["remaining"]) == "54.00" and c["clearings"][0]["seq"] == 1
assert str(must(api("GET", f"/bills/{b['bill_id']}", CLERK))["outstanding"]) == "162.00"
s, over = api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "55.00"})
assert s == 409 and "54.00 remains" in str(over)
run = must(propose())
assert str(run["payments"][0]["amount"]) == "162.00"
def test_pur49_a_zero_clearing_needs_a_reason_and_releases_the_bill(neighbours):
o, d, pid = paid_dp(neighbours)
receive(o, quantity="10")
s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
assert api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "0"})[0] == 422
must(api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "0", "reason": "a later bill takes it"}))
assert str(must(propose())["payments"][0]["amount"]) == "216.00"
def test_pur49_two_clerks_on_one_snapshot_clear_it_once():
import pytest
from erp_platform.repository import Conflict
from test_stories_p2p import base_engine
from brick_purchasing.domain import rows
from brick_purchasing.infrastructure.tables import Repository
world()
with base_engine() as e:
c1, c2 = e.connect(), e.connect()
t1, t2 = c1.begin(), c2.begin()
r1, _r2 = Repository(c1), Repository(c2)
d = rows.DownPayment("dp:X:1", "DP-1", "X", 7001, 100, "", 0, 100, "GBP", None, "paid")
t0 = e.connect()
t0.close()
r1.down_payments.save(d, 0)
t1.commit()
t2.rollback()
c3, c4 = e.connect(), e.connect()
a, b = c3.begin(), c4.begin()
x, v = Repository(c3).down_payments.get("dp:X:1")
y, w = Repository(c4).down_payments.get("dp:X:1")
x.clearings.append(rows.DownPaymentClearing("dp:X:1", "B1", 1, 60))
y.clearings.append(rows.DownPaymentClearing("dp:X:1", "B1", 1, 60))
Repository(c3).down_payments.save(x, v)
a.commit()
with pytest.raises(Conflict):
Repository(c4).down_payments.save(y, w)
b.rollback()
# ── the VAT on a down payment (PUR-48, PUR-49 second half) ───────────────────────────────────────────
def test_pur47_down_payment_vat_is_computed_by_tax_and_the_stated_amount_only_a_fallback(neighbours, monkeypatch):
o, d = order_with_dp()
assert str(d["tax_amount"]) == "18.00" and neighbours.tax.docs[f"pur_dp:{d['down_payment_id']}"]["direction"] == "purchase"
o2 = confirmed(lines=[{"product_id": "COLA-075", "quantity": "10"}])
wrong = must(api("POST", f"/purchase-orders/{o2['purchase_order_id']}/down-payments", BUYER, {"percent": "50", "due_date": DUE, "tax_amount": "5.00"}))
assert str(wrong["tax_amount"]) == "18.00" # an input never overrides tax
monkeypatch.setattr(type(neighbours.tax), "compute", lambda self, body: (_ for _ in ()).throw(AssertionError("tax is not installed")))
real = seams_installed(monkeypatch)
o3 = confirmed(lines=[{"product_id": "COLA-075", "quantity": "10"}])
fb = must(api("POST", f"/purchase-orders/{o3['purchase_order_id']}/down-payments", BUYER, {"percent": "50", "due_date": DUE, "tax_amount": "5.00"}))
assert str(fb["tax_amount"]) == "5.00" and str(fb["gross_amount"]) == "95.00" and real
def seams_installed(monkeypatch):
from brick_purchasing.infrastructure import seams
orig = seams.ports.installed
monkeypatch.setattr(type(seams.ports), "installed", lambda self, name: False if name == "tax" else orig(name))
return True
def invoiced(neighbours):
o, d, pid = paid_dp(neighbours)
d = must(api("POST", f"/down-payments/{d['down_payment_id']}/invoice", CLERK, {"supplier_invoice_no": "DPI-1", "invoice_date": "2026-11-20"}))
inv = must(api("GET", f"/bills/{d['invoice_bill_id']}", CLERK))
return o, d, inv
def test_pur48_the_vat_invoice_moves_only_the_vat_at_approval_and_is_never_paid(neighbours):
o, d, inv = invoiced(neighbours)
assert inv["kind"] == "down_payment_invoice" and str(inv["tax_total"]) == "18.00" and inv["tax_ref"]
again = must(api("POST", f"/down-payments/{d['down_payment_id']}/invoice", CLERK, {"supplier_invoice_no": "DPI-1", "invoice_date": "2026-11-20"}))
assert again["invoice_bill_id"] == d["invoice_bill_id"]
assert neighbours.gl.posted(d["invoice_bill_id"]) == [] # nothing before approval
must(api("POST", f"/bills/{d['invoice_bill_id']}/approve", PAYAPPROVER, {}))
(j,) = neighbours.gl.posted(d["invoice_bill_id"], "approve")
assert sorted((x["account_id"], x.get("debit") or x.get("credit"), "d" if x.get("debit") else "c") for x in j["lines"]) == [("1600", "18.00", "c"), ("2100", "18.00", "d")] \
or {x["account_id"] for x in j["lines"]} >= {"1600"}
assert d["invoice_bill_id"] in neighbours.tax.frozen or f"pur_bill:{d['invoice_bill_id']}" in neighbours.tax.frozen
s, r = propose() # the invoice is not payable
assert s == 409 and "nothing falls due" in str(r)
def test_pur49_clearing_reverses_the_reclaimed_vat_and_posts_a_balanced_journal(neighbours):
o, d, inv = invoiced(neighbours)
must(api("POST", f"/bills/{d['invoice_bill_id']}/approve", PAYAPPROVER, {}))
receive(o, quantity="10")
s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
c = must(api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "54.00"}))
assert c["posted"] is True and c["clearings"][0]["tax_code"]
(j,) = neighbours.gl.posted(d["down_payment_id"], "pay")
deb = sum(float(x.get("debit", 0)) for x in j["lines"])
cred = sum(float(x.get("credit", 0)) for x in j["lines"])
assert deb == cred == 54.0 and sorted(x["credit"] for x in j["lines"] if "credit" in x) == ["45.00", "9.00"] # 9.00 = half of the 18.00 reclaimed
def test_pur49_a_refund_returns_what_is_left_and_a_left_over_down_payment_is_listed(neighbours):
o, d, inv = invoiced(neighbours)
must(api("POST", f"/bills/{d['invoice_bill_id']}/approve", PAYAPPROVER, {}))
must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/close-short", BUYER, {"reason": "supplier cannot deliver"}))
left = must(api("GET", "/down-payments?left_on_closed_orders=true", BUYER))["items"]
assert [x["down_payment_id"] for x in left] == [d["down_payment_id"]] and "refund" in left[0]["fix"]
assert api("POST", f"/down-payments/{d['down_payment_id']}/refund", CLERK, {"refund_reference": "RF-1", "amount": "108.01"})[0] == 409
r = must(api("POST", f"/down-payments/{d['down_payment_id']}/refund", CLERK, {"refund_reference": "RF-1"}))
assert str(r["remaining"]) == "0.00" or float(r["remaining"]) == 0.0
(j,) = neighbours.gl.posted(d["down_payment_id"], "reverse")
assert sorted(x["credit"] for x in j["lines"] if "credit" in x) == ["18.00", "90.00"]
again = must(api("POST", f"/down-payments/{d['down_payment_id']}/refund", CLERK, {"refund_reference": "RF-1"}))
assert len(neighbours.gl.posted(d["down_payment_id"], "reverse")) == 1 and len(again["clearings"]) == 1
assert must(api("GET", "/down-payments?left_on_closed_orders=true", BUYER))["items"] == []