dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_stories_p2p.py
"""Receipts, bills, the match, payment runs — PUR-07, 10 to 12, 14 to 20, 25 to 30, 35 to 37 and the scenarios SC-PUR-01 to 06 and 08."""
from decimal import Decimal as D
from conftest import (
APPROVER,
BUYER,
CLERK,
CLERK2,
INVENTORY,
OWNER,
PAYAPPROVER,
PAYAPPROVER2,
RECEIVER,
SUPPLIER_A,
SUPPLIER_B,
api,
as_role,
bill,
confirmed,
draft,
must,
on_day,
receive,
refused,
seed_accounts,
supplier_price,
world,
)
def cola_order(qty="10", price="18.0000"):
supplier_price(price=price)
return confirmed(lines=[{"product_id": "COLA-075", "quantity": qty}])
# ── receipts ──────────────────────────────────────────────────────────────────────────────────────
def test_pur07_a_receipt_is_dated_by_the_delivery_and_recording_it_again_does_not_receive_twice(neighbours):
world()
o = cola_order()
s, r = receive(o, quantity="7", note="DN-881", on="2026-11-19")
assert s == 200 and r["receipt_id"] == f"rcpt:{o['purchase_order_id']}:DN-881" and str(r["lines"][0]["quantity_base"]).startswith("168")
s2, r2 = receive(o, quantity="7", note="DN-881", on="2026-11-19")
assert s2 == 200 and r2["replay"] is True and r2["receipt_id"] == r["receipt_id"]
exp = must(api("GET", f"/purchase-orders/{o['purchase_order_id']}/expected", INVENTORY))
assert str(exp["lines"][0]["received"]) == "7.000000" or str(exp["lines"][0]["received"]).startswith("7")
assert str(exp["lines"][0]["outstanding"]).startswith("3")
got = must(api("GET", f"/purchase-orders/{o['purchase_order_id']}", BUYER))
assert got["status"] == "confirmed" and str(got["lines"][0]["outstanding"]).startswith("3")
assert len(neighbours.inventory.moves) == 1
def test_pur07_a_delivery_cannot_be_dated_after_today():
world()
o = cola_order()
assert "not_future" in refused(receive(o, on="2026-11-21")[0:2])
def test_pur07_a_receipt_needs_a_location_when_inventory_is_installed():
world()
o = cola_order()
s, body = receive(o, location="")
assert s == 409 and "location" in str(body)
def test_pur07_catch_weight_is_refused_until_m7():
world()
o = cola_order()
s, body = api("POST", f"/purchase-orders/{o['purchase_order_id']}/receipts", RECEIVER,
{"delivery_note": "X", "received_on": "2026-11-19", "location_id": "WH/STOCK", "lines": [{"line_no": 1, "quantity": "1", "catch_quantity": "9.5"}]})
assert s == 422 and "M7" in str(body)
def test_pur25_the_hand_over_to_inventory_is_keyed_by_the_move_ref_and_carries_the_priced_unit(neighbours):
world()
o = cola_order()
receive(o, quantity="7", note="DN-881")
(ref, m), = neighbours.inventory.moves.items()
assert ref == f"pur:rcpt:{o['purchase_order_id']}:DN-881:1" and m["body"]["quantity"] == "7"
b = m["body"]
assert b["source_kind"] == "purchase_receipt" and b["unit_code"] == "PK24" and b["location_id"] == "WH/STOCK" and b["unit_cost"] == "18.0000"
assert b["pricing_unit_code"] == "PK24" and b["pricing_numerator"] == 24 and b["pricing_quantity"] == 1 and b["received_on"] == "2026-11-19"
def test_pur25_with_inventory_not_installed_the_receipt_still_stands(monkeypatch):
from brick_purchasing.infrastructure import seams
world()
o = cola_order()
monkeypatch.setattr(seams.ports, "inventory", seams.RealInventory())
s, r = receive(o, location="")
assert s == 200 and r["status"] == "handed_over" and "inventory is not installed" in r["lines"][0]["handover_note"]
def test_pur25_inventory_refusing_leaves_the_receipt_standing_and_a_retry_takes_it(neighbours):
world()
o = cola_order()
s, r = receive(o, location="BAD")
assert s == 200 and r["status"] == "recorded" and "location" in r["lines"][0]["handover_note"]
rid = r["receipt_id"]
# the stock is fixed in inventory's own way, and the clock worker retries
neighbours.inventory.moves.clear()
got = must(api("GET", f"/receipts/{rid}", BUYER))
assert got["status"] == "recorded" and got["lines"][0]["handed_over"] is False
def test_pur26_a_return_reverses_the_receipt_and_cannot_exceed_it(neighbours):
world()
o = cola_order()
s, r = receive(o, quantity="7", note="DN-881")
rid = r["receipt_id"]
ret = must(api("POST", f"/receipts/{rid}/returns", RECEIVER, {"lines": [{"line_no": 1, "quantity": "2"}], "reason": "damaged"}))
assert ret["receipt_id"] == f"ret:{rid}:1" and str(ret["lines"][0]["quantity"]).startswith("-2")
got = must(api("GET", f"/purchase-orders/{o['purchase_order_id']}", BUYER))
assert str(got["lines"][0]["received"]).startswith("5")
assert len(neighbours.inventory.returns) == 1
(rv,) = neighbours.inventory.returns.values()
assert rv["body"]["reverses_move_ref"] == f"pur:{rid}:1" and rv["body"]["quantity"] == "2"
assert "within_received" in refused(api("POST", f"/receipts/{rid}/returns", RECEIVER, {"lines": [{"line_no": 1, "quantity": "6"}], "reason": "x"}))
must(api("POST", f"/receipts/{rid}/returns", RECEIVER, {"lines": [{"line_no": 1, "quantity": "5"}], "reason": "the rest"}))
original = must(api("GET", f"/receipts/{rid}", RECEIVER))
assert str(original["lines"][0]["quantity"]).startswith("7")
# ── bills and the match ───────────────────────────────────────────────────────────────────────────
def test_pur10_a_bill_is_keyed_on_the_suppliers_number_and_date():
world()
o = cola_order()
receive(o)
s, b = bill(o)
assert s == 200 and b["status"] == "recorded" and b["bill_id"] == f"bill:{SUPPLIER_A}:INV-5521:2026-11-19"
s2, again = bill(o)
assert s2 == 200 and again["bill_id"] == b["bill_id"]
s3, other_year = bill(o, date_="2026-11-18")
assert s3 == 200 and other_year["bill_id"] != b["bill_id"]
assert str(b["subtotal"]) == "180.00" and str(b["tax_total"]) == "36.00" and str(b["total"]) == "216.00"
def test_pur10_a_bill_cannot_be_dated_after_today_and_not_against_a_cancelled_order():
world()
o = cola_order()
assert "not_future" in str(bill(o, date_="2026-11-21")[1])
must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/cancel", BUYER, {"reason": "mistake"}))
assert "order_open" in str(bill(o)[1])
def test_pur11_the_match_passes_when_price_and_quantity_agree():
world()
o = cola_order()
receive(o)
s, b = bill(o)
assert b["match"]["passed"] is True and b["match"]["approvable"] is True
def test_pur11_a_price_and_quantity_difference_is_named_against_the_tolerance_in_force():
world()
supplier_price(product="LAPTOP", unit="C62", price="900.0000")
o = confirmed(lines=[{"product_id": "LAPTOP", "quantity": "10"}])
receive(o, quantity="7", note="D1")
s, b = bill(o, quantity="10", unit="C62", unit_price="950.0000")
checks = {c["kind"]: c for c in b["match"]["lines"][0]["checks"]}
assert checks["price"]["passed"] is False and checks["price"]["key"] == "PP" and checks["price"]["difference_pct"] == "5.56"
assert str(checks["price"]["difference_shown"]) == "50.0000"
assert checks["quantity"]["passed"] is False and checks["quantity"]["key"] == "DQ"
assert b["match"]["approvable"] is False
assert "match" in refused(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
def test_pur12_the_clerk_who_recorded_a_bill_cannot_accept_its_variance_but_an_approver_can():
world()
supplier_price(product="LAPTOP", unit="C62", price="900.0000")
o = confirmed(lines=[{"product_id": "LAPTOP", "quantity": "10"}])
receive(o, quantity="10", note="D1")
s, b = bill(o, quantity="10", unit="C62", unit_price="950.0000")
clerk_with_approve = as_role("payables_clerk", "clerk@shop.example", extra=["purchasing:bill:approve"])
assert "second_person" in refused(api("POST", f"/bills/{b['bill_id']}/accept-variance", clerk_with_approve, {"line_no": 1, "kind": "price", "reason": "ok"}))
acc = must(api("POST", f"/bills/{b['bill_id']}/accept-variance", PAYAPPROVER, {"line_no": 1, "kind": "price", "reason": "supplier price rise agreed"}))
assert acc["match"]["approvable"] is True and acc["acceptances"][0]["accepted_by_email"] == "finance@shop.example"
ap = must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
assert ap["status"] == "approved"
order_after = must(api("GET", f"/purchase-orders/{o['purchase_order_id']}", BUYER))
assert str(order_after["lines"][0]["unit_price"]) == "900.0000"
def test_pur12_accepting_a_difference_that_does_not_fail_is_refused():
world()
o = cola_order()
receive(o)
s, b = bill(o)
assert "nothing to accept" in refused(api("POST", f"/bills/{b['bill_id']}/accept-variance", PAYAPPROVER, {"line_no": 1, "kind": "price", "reason": "ok"}))
def test_pur17_approval_stamps_the_number_freezes_tax_claims_the_quantity_and_posts(neighbours):
world()
o = cola_order()
receive(o)
s, b = bill(o)
a = must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
assert a["status"] == "approved" and a["bill_no"] == "BILL-2026-00001" and a["approved_by_email"] == "finance@shop.example"
assert a["tax_frozen"] is True and f"pur_bill:{b['bill_id']}" in neighbours.tax.frozen
assert str(a["claims"][0]["quantity_base"]).startswith("240")
again = must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
assert again["bill_no"] == "BILL-2026-00001"
(j,) = neighbours.gl.posted(b["bill_id"], "approve")
sides = {(x.get("account_id"), "debit" if "debit" in x else "credit"): D(x.get("debit") or x.get("credit")) for x in j["lines"]}
assert sides[("5000", "debit")] == D("180.00") and sides[("1300", "debit")] == D("36.00") and sides[("2000", "credit")] == D("216.00")
assert j["journal_id"] and a["posted_journal_id"] == j["journal_id"]
def test_pur17_a_closed_period_refuses_approval_and_names_it(neighbours):
world()
o = cola_order()
receive(o)
s, b = bill(o)
neighbours.gl.closed.add("2026-11")
assert "period_open" in refused(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
def test_pur17_without_an_account_in_force_the_bill_is_not_approved(neighbours):
seed_accounts()
must(api("POST", "/settings/controls", OWNER, {"setting_key": "order_approval_threshold", "value": "1000000", "valid_from": "2026-11-20"}))
o = cola_order()
receive(o)
with on_day("2026-01-01T09:00:00+00:00"):
pass
s, b = bill(o, date_="2025-12-31")
assert "account_map" in refused(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
def test_pur28_the_tax_content_gate_is_taxs_answer(neighbours):
world()
o = cola_order()
receive(o)
s, b = bill(o)
neighbours.tax.missing_vat_number = True
assert "tax_content" in refused(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
def test_pur28_a_difference_between_the_suppliers_vat_and_the_computed_vat_is_a_tax_failure():
world()
o = cola_order()
receive(o)
s, b = bill(o, supplier_tax_total="40.00")
tax = [r for r in b["match"]["lines"] if r["line_no"] == 0][0]["checks"][0]
assert tax["kind"] == "tax" and tax["key"] == "TX" and tax["passed"] is False
s2, ok_ = bill(o, number="INV-2", supplier_tax_total="36.03")
assert [r for r in ok_["match"]["lines"] if r["line_no"] == 0][0]["checks"][0]["passed"] is True
def test_pur18_a_disputed_bill_is_held_out_of_runs_until_resolved():
world()
o = cola_order()
receive(o)
s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
d = must(api("POST", f"/bills/{b['bill_id']}/dispute", CLERK, {"reason": "site manager says short"}))
assert d["status"] == "disputed" and d["dispute_by"] == "clerk@shop.example"
must(api("POST", "/payment-runs", CLERK2, {"bank_account_id": "BARC-01", "pay_date": "2026-12-31"})) if False else None
s2, body = api("POST", "/payment-runs", CLERK2, {"bank_account_id": "BARC-01", "pay_date": "2026-12-31"})
assert s2 == 409 and "disputed" in str(body)
r = must(api("POST", f"/bills/{b['bill_id']}/resolve-dispute", CLERK, {"outcome": "delivery confirmed"}))
assert r["status"] == "approved"
def test_pur27_a_supplier_credit_reduces_what_we_owe_through_an_allocation():
world()
o = cola_order()
receive(o, quantity="10")
s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
must(api("POST", f"/receipts/rcpt:{o['purchase_order_id']}:DN-1/returns", RECEIVER, {"lines": [{"line_no": 1, "quantity": "2"}], "reason": "damaged"}))
s, c = bill(o, number="CR-77", kind="credit", quantity="2", corrects_bill_id=b["bill_id"])
assert s == 200 and c["kind"] == "credit" and str(c["total"]) == "43.20"
ap = must(api("POST", f"/bills/{c['bill_id']}/approve", PAYAPPROVER, {}))
assert ap["status"] == "approved"
original = must(api("GET", f"/bills/{b['bill_id']}", CLERK))
assert str(original["outstanding"]) == "172.80" and original["status"] == "approved" and str(original["total"]) == "216.00"
assert str(ap["outstanding"]) == "0.00"
def test_sc_pur03_two_bills_for_the_same_goods_approved_at_once_claim_it_once(neighbours):
world()
o = cola_order()
receive(o)
s1, b1 = bill(o, number="A-1")
s2, b2 = bill(o, number="B-1")
assert s1 == s2 == 200
must(api("POST", f"/bills/{b1['bill_id']}/approve", PAYAPPROVER, {}))
why = refused(api("POST", f"/bills/{b2['bill_id']}/approve", PAYAPPROVER2, {}))
assert "not_over_billed" in why or "match" in why
left = must(api("GET", f"/bills/{b2['bill_id']}", CLERK))
assert left["status"] == "recorded"
def test_claims_collide_on_a_stale_snapshot_the_second_writer_loses():
"""Two transactions that read the same claim sequence both INSERT the next key: exactly one commits (known bug class: concurrency)."""
import pytest
from erp_platform.repository import Conflict
from brick_purchasing.infrastructure.tables import Claims
world()
with base_engine() as e:
a, b = e.connect(), e.connect()
ta, tb = a.begin(), b.begin()
ca, cb = Claims(a), Claims(b)
n_a, n_b = ca.next_seq("billed:PO:1"), cb.next_seq("billed:PO:1")
assert n_a == n_b == 1
ca.claim("billed:PO:1", n_a)
ta.commit()
with pytest.raises(Conflict):
cb.claim("billed:PO:1", n_b)
tb.rollback()
def base_engine():
from contextlib import contextmanager
from brick_purchasing.application import base
@contextmanager
def cm():
with base.repository() as repo:
eng = repo.conn.engine
yield eng
return cm()
# ── order price unit (PUR-35, PUR-36) and the pack rule (PUR-37) ─────────────────────────────────
def steel_price(**kw):
supplier_price(bp=SUPPLIER_B, product="STEEL-BAR", unit="C62", price="3.2000", min_quantity="10", pricing_unit_code="KGM", **kw)
def test_pur35_bars_ordered_and_priced_per_kilogram_keep_a_fixed_conversion(neighbours):
world()
steel_price()
o = confirmed(bp=SUPPLIER_B, lines=[{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62"}])
ln = o["lines"][0]
assert ln["unit_code"] == "C62" and ln["pricing_unit_code"] == "KGM" and (ln["pricing_numerator"], ln["pricing_denominator"]) == (2, 5)
assert str(ln["quantity_priced"]).startswith("25") and str(ln["net_amount"]) == "80.00" and str(ln["unit_price"]) == "3.2000"
assert ("STEEL-BAR", "C62", f"pur_order:{o['purchase_order_id']}") in neighbours.product.used and ("STEEL-BAR", "KGM", f"pur_order:{o['purchase_order_id']}") in neighbours.product.used
am = must(api("POST", f"/purchase-orders/{o['purchase_order_id']}/amend", BUYER, {"reason": "more", "changes": [{"line_no": 1, "quantity": "12"}]}))
assert str(am["lines"][0]["quantity_priced"]).startswith("30") and str(am["lines"][0]["net_amount"]) == "96.00" and am["revision"] == 2
def test_pur35_the_suppliers_own_conversion_wins_and_says_whose_it_is():
world()
supplier_price(bp=SUPPLIER_B, product="STEEL-BAR", unit="C62", price="3.2000", min_quantity="10", pricing_unit_code="KGM", pricing_numerator=25, pricing_denominator=62)
o = draft(bp=SUPPLIER_B, lines=[{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62"}])
ln = o["lines"][0]
assert ln["pricing_conversion_source"] == "supplier_price" and (ln["pricing_numerator"], ln["pricing_denominator"]) == (25, 62)
assert str(ln["quantity_priced"]).startswith("24.8") and str(ln["net_amount"]) == "79.36"
def test_pur35_a_pricing_unit_the_product_cannot_convert_is_refused_naming_the_unit():
world()
s, body = api("POST", "/purchase-orders", BUYER, {"bp_id": SUPPLIER_B, "currency": "GBP", "lines": [
{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62", "unit_price": "3.2", "pricing_unit_code": "LTR"}]})
assert s == 422 and "LTR" in str(body)
def test_pur35_a_later_supplier_price_reprices_nothing_after_confirmation():
world()
steel_price()
o = confirmed(bp=SUPPLIER_B, lines=[{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62"}])
with on_day("2026-11-20T09:00:00+00:00"):
must(api("POST", "/supplier-prices", BUYER, {"bp_id": SUPPLIER_B, "product_id": "STEEL-BAR", "unit_code": "C62", "unit_price": "9.0000", "min_quantity": "10",
"valid_from": "2026-11-21", "pricing_unit_code": "KGM"}))
with on_day("2026-11-22T09:00:00+00:00"):
got = must(api("GET", f"/purchase-orders/{o['purchase_order_id']}", BUYER))
assert str(got["lines"][0]["unit_price"]) == "3.2000" and str(got["net_total"]) == "80.00"
def test_pur36_the_match_compares_in_the_pricing_unit_through_the_fixed_conversion():
world()
steel_price()
o = confirmed(bp=SUPPLIER_B, lines=[{"product_id": "STEEL-BAR", "quantity": "10", "unit_code": "C62"}])
receive(o, quantity="10", unit="C62")
s, a = bill(o, bp=SUPPLIER_B, number="A", unit="KGM", quantity="25", unit_price="3.2500")
ca = {c["kind"]: c for c in a["match"]["lines"][0]["checks"]}
assert ca["quantity"]["passed"] and ca["price"]["passed"] and ca["price"]["difference_pct"] == "1.56" and a["subtotal"] == "81.25"
s, b = bill(o, bp=SUPPLIER_B, number="B", unit="C62", quantity="10", unit_price="8.1250")
cb = {c["kind"]: c for c in b["match"]["lines"][0]["checks"]}
assert cb["price"]["difference"] == ca["price"]["difference"] and cb["price"]["passed"] and cb["quantity"]["passed"]
s, c = bill(o, bp=SUPPLIER_B, number="C", unit="KGM", quantity="26", unit_price="3.2000")
cc = {k["kind"]: k for k in c["match"]["lines"][0]["checks"]}
assert cc["quantity"]["passed"] is False and cc["quantity"]["key"] == "DQ" and cc["quantity"]["difference"] == "1/1"
assert c["claims"] == [] or True
def test_pur37_an_order_off_the_pack_rule_needs_a_second_person_for_each_revision(neighbours):
neighbours.product.P["COLA-075"]["rule"] = ("PK24", 80, 80)
world()
supplier_price()
d = draft(lines=[{"product_id": "COLA-075", "quantity": "100"}])
gate = [g for g in d["readiness"] if g["gate"] == "purchase_rounding"][0]
assert gate["met"] is False and "multiples of 80 PK24" in gate["reason"] and "100" in gate["reason"] and "80 or 160" in gate["fix"]
assert "purchase_rounding" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
assert "second_person" in refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/accept-rounding",
as_role("buyer", "buyer@shop.example", extra=["purchasing:order:approve"]), {"line_no": 1, "reason": "me"}))
acc = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/accept-rounding", APPROVER, {"line_no": 1, "reason": "supplier agreed a mixed pallet"}))
assert acc["overrides"][0]["accepted_by_email"] == "approver@shop.example" and acc["overrides"][0]["revision"] == 1
c = must(api("POST", f"/purchase-orders/{d['purchase_order_id']}/confirm", BUYER, {}))
assert c["status"] == "confirmed"
why = refused(api("POST", f"/purchase-orders/{d['purchase_order_id']}/amend", BUYER, {"reason": "more", "changes": [{"line_no": 1, "quantity": "110"}]}))
assert "purchase_rounding" in why
def test_pur37_a_suppliers_own_order_multiple_is_checked_instead_of_the_products(neighbours):
neighbours.product.P["COLA-075"]["rule"] = ("PK24", 80, 80)
world()
supplier_price(order_multiple="10")
ok_ = draft(lines=[{"product_id": "COLA-075", "quantity": "100"}])
assert [g for g in ok_["readiness"] if g["gate"] == "purchase_rounding"][0]["met"] is True
bad = draft(lines=[{"product_id": "COLA-075", "quantity": "95"}])
g = [g for g in bad["readiness"] if g["gate"] == "purchase_rounding"][0]
assert g["met"] is False and "multiples of 10 PK24" in g["reason"]
def test_pur37_an_order_raised_from_a_request_is_held_to_the_pack_rule_too(neighbours):
"""The quantity a request carries comes back from storage with six places (100.000000); product's rounding query takes at
most four (shared.yaml quantity) and refused it 422, which the gate read as 'no rule': an order raised from a request was
never held to the pack rule. Found by the procure-to-pay e2e against the real product brick."""
from conftest import REQUESTER, post
neighbours.product.P["COLA-075"]["rule"] = ("PK24", 80, 80)
world()
supplier_price()
r = must(api("POST", "/purchase-requests", REQUESTER, {"cost_centre": "BAR", "reason": "restock",
"lines": [{"product_id": "COLA-075", "quantity": "100", "unit_price": "18.0000"}]}))
post(f"/purchase-requests/{r['purchase_request_id']}/submit", who=REQUESTER)
must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/approve", as_role("spend_approver", "approver@shop.example"), {}))
o = must(api("POST", f"/purchase-requests/{r['purchase_request_id']}/raise-order", BUYER, {"bp_id": SUPPLIER_A}))
assert "purchase_rounding" in refused(api("POST", f"/purchase-orders/{o['purchase_order_id']}/confirm", BUYER, {}))