dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_audit_musts.py
"""Audit MUSTs (wave 2, purchasing @7f8a070): each test FAILS on the audited code for the reason in its docstring."""
import pytest
from conftest import (
ALICE,
BUYER,
CLERK,
INVENTORY,
OWNER,
PAYAPPROVER,
PAYAPPROVER2,
REQUESTER,
SUPPLIER_A,
api,
bill,
confirmed,
must,
person,
receive,
supplier_price,
world,
)
from erp_platform.repository import Conflict
from erp_platform.seam import derived_id
from test_stories_assets_consign import consigned
from test_stories_downpayments import paid_dp
from test_stories_migration import odoo
from brick_purchasing.application import base
from brick_purchasing.infrastructure import seams
from brick_purchasing.infrastructure.tables import Claims
def _cleared(neighbours, amount="54.00"):
o, d, _pid = paid_dp(neighbours)
receive(o, quantity="10")
_s, b = bill(o)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
return o, d, b
def test_m1_down_payment_clearing_takes_the_bills_allocation_claim(neighbours):
"""Class 4: clear_down_payment lowers a bill's outstanding without claiming `alloc:<bill_id>` (payments and credits do), so a payment
allocation on a snapshot taken before the clearing commits too: the bill is settled twice."""
_o, d, b = _cleared(neighbours)
with base.repository() as repo:
e = repo.conn.engine
stale = Claims(repo.conn).next_seq(f"alloc:{b['bill_id']}") # what a concurrent payment read before the clearing committed
must(api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": b["bill_id"], "amount": "54.00"}))
c = e.connect()
t = c.begin()
try:
with pytest.raises(Conflict):
Claims(c).claim(f"alloc:{b['bill_id']}", stale)
finally:
t.rollback()
c.close()
def test_m2_down_payment_clearing_journal_matches_gls_contract(neighbours):
"""Class 8: gl's purchasing seam takes event approve | pay | reverse and journal_id = derived_id(source_document, event), checked by gl;
the clearing posts event 'clear' with journal_id derived from '<dp>:<bill>:<seq>' while source_document is the down payment, so the real
ledger refuses it (and a second clearing of the same down payment is a different body under the same derived id)."""
_o, d, _b = _cleared(neighbours)
neighbours.gl.journals.clear()
must(api("POST", f"/down-payments/{d['down_payment_id']}/clear", CLERK, {"bill_id": _b["bill_id"], "amount": "54.00"}))
(j,) = neighbours.gl.posted(d["down_payment_id"])
assert j["event"] in ("approve", "pay", "reverse"), j["event"]
assert j["journal_id"] == str(derived_id(j["source_document"], j["event"])), "journal_id is not derived_id(source_document, event)"
def test_m3_a_settled_consignment_withdrawal_cannot_be_billed_again(neighbours):
"""Class 4: a withdrawal bounds what is billed for it, but match_withdrawals ignores its status and _claim_lines skips withdrawal lines, so a
second supplier invoice for the same withdrawal matches and is approved: the supplier is paid twice for one withdrawal."""
o = consigned()
receive(o, quantity="50")
must(api("POST", "/consignment-withdrawals", INVENTORY, {"move_ref": "inv:M77", "bp_id": SUPPLIER_A, "product_id": "COLA-075", "quantity": "18",
"unit_code": "PK24", "withdrawn_on": "2026-11-12"}))
wl = [{"order_line_no": 1, "withdrawal_id": "wd:inv:M77", "quantity": "18", "unit_code": "PK24", "unit_price": "18.0000"}]
_s, b1 = bill(o, number="INV-A", lines=wl)
must(api("POST", f"/bills/{b1['bill_id']}/approve", PAYAPPROVER, {}))
s, b2 = bill(o, number="INV-B", lines=wl)
if s == 200:
s, b2 = api("POST", f"/bills/{b2['bill_id']}/approve", PAYAPPROVER2, {})
assert s != 200, ("the same withdrawal was billed and approved twice", b2)
def test_m4_new_order_numbers_continue_above_the_migrated_series(neighbours):
"""Class 6: spec migration 'the highest purchase.order.name per year -> number_series.high_water_mark, so the next number continues the
carried series'; load_odoo only claims the rendered name and never raises the mark, so the next order fills the gap below P00007."""
world()
must(api("POST", "/settings/controls", OWNER, {"setting_key": "order_number_format", "value": "P{seq:5}", "valid_from": "2026-11-20"}))
odoo(neighbours)
must(api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"}))
supplier_price()
o = confirmed(lines=[{"product_id": "COLA-075", "quantity": "10"}])
assert o["order_no"] == "P00008", o["order_no"]
def test_m5_migration_load_without_an_odoo_reader_is_refused_not_a_crash(neighbours, monkeypatch):
"""Class 8: Ports() leaves odoo and maps None and nothing wires a real reader or product's GET /migration-map; load_odoo dereferences
ports.odoo.db, so in production the door answers a 500 (AttributeError) instead of a clean refusal."""
world()
monkeypatch.setattr(seams.ports, "odoo", None)
monkeypatch.setattr(seams.ports, "maps", None)
try:
s, b = api("POST", "/migration/odoo/load", ALICE, {"cutover_date": "2026-06-30"})
except AttributeError as e:
pytest.fail(f"500: {e}")
assert s in (409, 422, 503), (s, b)
def test_m6_an_order_line_whose_net_overflows_numeric_is_a_422(neighbours):
"""Class 5: price (<= 15 digits) x quantity (<= 13 digits) is never bounded, so the line net and order total overflow NUMERIC(18,2) on the
real store (a 500); the audited code accepts it."""
world()
s, b = api("POST", "/purchase-orders", BUYER, {"bp_id": SUPPLIER_A, "currency": "GBP",
"lines": [{"product_id": "COLA-075", "quantity": "9999999999999", "unit_price": "999999999999999"}]})
assert s == 422, (s, b)
def test_clean_requester_with_no_email_sees_only_their_own(neighbours):
"""Class 2 (control, expected to PASS): a requester with no email is scoped by sub."""
world()
nomail = person("", ["purchasing:request:read_own", "purchasing:request:write"], sub="sub-77")
r = must(api("POST", "/purchase-requests", REQUESTER, {"cost_centre": "BAR", "reason": "restock", "lines": [{"product_id": "COLA-075", "quantity": "10", "unit_price": "18.0000"}]}))
assert must(api("GET", "/purchase-requests", nomail, None))["total"] == 0
assert api("GET", f"/purchase-requests/{r['purchase_request_id']}", nomail, None)[0] == 404