dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_stories_assets_consign.py
"""Consignment (PUR-53, PUR-54), account assignment and asset hand-over (PUR-55, PUR-56), subcontracting (PUR-57, PUR-58), non-valuated receipts."""
from conftest import (
ALICE,
BUYER,
CLERK,
INVENTORY,
PAYAPPROVER,
SUPPLIER_A,
SYSTEM,
api,
bill,
confirmed,
draft,
must,
on_day,
receive,
supplier_price,
world,
)
VAN = "UK01-2026-VEH-0003-0"
def seed(purpose, account, cat="*"):
with on_day("2026-01-01T09:00:00+00:00"):
must(api("POST", "/account-map", ALICE, {"purpose": purpose, "category": cat, "account_id": account, "valid_from": "2026-01-02"}))
# ── consignment ───────────────────────────────────────────────────────────────────────────────────
def consigned():
world()
seed("consignment_clearing", "2150")
supplier_price(consignment=True)
return confirmed(lines=[{"product_id": "COLA-075", "quantity": "50", "item_category": "consignment"}])
def test_pur53_a_consignment_receipt_is_handed_over_as_the_suppliers_stock(neighbours):
o = consigned()
s, r = receive(o, quantity="50")
assert s == 200
(m,) = neighbours.inventory.moves.values()
assert m["body"]["stock_owner_bp_id"] == str(SUPPLIER_A) and m["body"]["source_kind"] == "consignment_receipt" and "unit_cost" not in m["body"]
s, b = bill(o) # no bill can match a consignment receipt
assert s == 422 and "withdrawals" in str(b)
def test_pur54_a_withdrawal_is_priced_on_the_day_idempotent_and_a_missing_price_is_refused(neighbours):
consigned()
body = {"move_ref": "inv:M77", "bp_id": SUPPLIER_A, "product_id": "COLA-075", "quantity": "18", "unit_code": "PK24", "withdrawn_on": "2026-11-12"}
w = must(api("POST", "/consignment-withdrawals", INVENTORY, body))
assert str(w["amount"]) == "324.00" and str(w["unit_price"]) == "18.0000" and w["withdrawal_id"] == "wd:inv:M77"
again = must(api("POST", "/consignment-withdrawals", INVENTORY, body))
assert again["replay"] is True and len(must(api("GET", "/consignment-withdrawals", CLERK))["items"]) == 1
s, b = api("POST", "/consignment-withdrawals", INVENTORY, {**body, "move_ref": "inv:M78", "product_id": "STEEL-BAR", "unit_code": "KGM"})
assert s == 409 and "consignment_price" in str(b)
assert api("POST", "/consignment-withdrawals", INVENTORY, {**body, "move_ref": "inv:M79", "withdrawn_on": "2026-12-25"})[0] == 409
assert api("POST", "/consignment-withdrawals", BUYER, body)[0] == 403
rev = must(api("POST", "/consignment-withdrawals", INVENTORY, {**body, "move_ref": "inv:M80", "reverses_move_ref": "inv:M77"}))
assert str(rev["amount"]) == "-324.00"
def test_pur54_the_suppliers_invoice_line_is_matched_against_the_withdrawal_and_settles_it(neighbours):
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"}))
s, b = bill(o, lines=[{"order_line_no": 1, "withdrawal_id": "wd:inv:M77", "quantity": "18", "unit_code": "PK24", "unit_price": "18.0000"}])
assert s == 200, b
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
(j,) = neighbours.gl.posted(b["bill_id"], "approve")
assert any(x.get("purpose") == "consignment_clearing" for x in j["lines"])
assert must(api("GET", "/consignment-withdrawals?open_only=false", CLERK))["items"][0]["settled_bill_id"] == b["bill_id"]
# ── account assignment and the asset hand-over ────────────────────────────────────────────────────
def van_order(**kw):
world()
seed("asset_clearing", "0900")
line = {"product_id": "LAPTOP", "quantity": "1", "unit_price": "32000.00", "account_assignment": "asset", "asset_id": VAN, "cost_centre": "SITE-A", "company_code": "UK01", **kw}
return confirmed(lines=[line])
def test_pur55_the_asset_gates_refuse_a_retired_asset_another_company_or_an_unknown_one():
world()
for asset, why in (("UK01-2026-VEH-0009-0", "retired"), ("UK02-2026-VEH-0001-0", "company"), ("NOPE", "no asset")):
o = draft(lines=[{"product_id": "LAPTOP", "quantity": "1", "unit_price": "900.00", "account_assignment": "asset", "asset_id": asset, "company_code": "UK01"}])
s, b = api("POST", f"/purchase-orders/{o['purchase_order_id']}/confirm", BUYER, {})
assert s == 409 and "asset_assignable" in str(b), (asset, why, b)
def test_pur55_an_asset_line_is_non_valuated_and_a_cost_centre_line_is_expensed_with_its_centre(neighbours):
o = van_order()
receive(o, quantity="1")
assert neighbours.inventory.moves == {} # no stock, no GRNI
s, b = bill(o, lines=[{"order_line_no": 1, "quantity": "1", "unit_code": "C62", "unit_price": "32000.00"}])
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
(j,) = neighbours.gl.posted(b["bill_id"], "approve")
assert any(x.get("purpose") == "asset_clearing" and x.get("asset_id") == VAN for x in j["lines"])
o2 = confirmed(lines=[{"product_id": "CONSULT", "quantity": "2", "unit_price": "100.00", "account_assignment": "cost_centre", "cost_centre": "SITE-A"}])
receive(o2, quantity="2", note="DN-9")
s, b2 = bill(o2, number="INV-9", lines=[{"order_line_no": 1, "quantity": "2", "unit_code": "HUR", "unit_price": "100.00"}])
must(api("POST", f"/bills/{b2['bill_id']}/approve", PAYAPPROVER, {}))
(j2,) = neighbours.gl.posted(b2["bill_id"], "approve")
assert any(x.get("cost_centre") == "SITE-A" for x in j2["lines"])
def van_bill(neighbours):
o = van_order()
receive(o, quantity="1")
s, b = bill(o, lines=[{"order_line_no": 1, "quantity": "1", "unit_code": "C62", "unit_price": "32000.00"}])
return o, b
def test_pur56_approval_hands_the_asset_line_over_with_net_plus_nonreclaimable_vat_and_a_retry_posts_once(neighbours):
o, b = van_bill(neighbours)
must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
ref = f"pur:{b['bill_id']}:1"
sent = neighbours.fa.acquisitions[ref]
assert sent["amount"] == "32000.00" and sent["asset_id"] == VAN and sent["source_kind"] == "purchase_bill" and sent["value_date"] == "2026-11-19"
got = must(api("GET", f"/bills/{b['bill_id']}", CLERK))
assert got["lines"][0]["fa_transaction_id"] == "T1"
again = must(api("POST", f"/bills/{b['bill_id']}/hand-over-assets", SYSTEM, {}))
assert len(neighbours.fa.acquisitions) == 1 and again["handovers"][0]["status"] == "done"
def test_pur56_a_refusal_keeps_the_bill_approved_lists_the_handover_and_the_retry_completes_it(neighbours):
o, b = van_bill(neighbours)
neighbours.fa.refuse = "the period is closed"
out = must(api("POST", f"/bills/{b['bill_id']}/approve", PAYAPPROVER, {}))
assert out["status"] == "approved"
(h,) = must(api("POST", f"/bills/{b['bill_id']}/hand-over-assets", SYSTEM, {}))["handovers"]
assert h["status"] == "pending" and "period is closed" in h["reason"] and h["attempts"] == 2
neighbours.fa.refuse = None
(h,) = must(api("POST", f"/bills/{b['bill_id']}/hand-over-assets", SYSTEM, {}))["handovers"]
assert h["status"] == "done" and len(neighbours.fa.acquisitions) == 1
assert api("POST", f"/bills/{b['bill_id']}/hand-over-assets", CLERK, {})[0] == 403
# ── subcontracting ────────────────────────────────────────────────────────────────────────────────
def subcontract_order():
world()
supplier_price()
comps = [{"product_id": "STEEL-BAR", "quantity_per": "1", "unit_code": "C62"}, {"product_id": "WIDGET", "quantity_per": "0.2", "unit_code": "C62"}]
return confirmed(lines=[{"product_id": "COLA-075", "quantity": "10", "item_category": "subcontract", "components": comps}])
def test_pur57_providing_moves_the_components_to_the_subcontractor_and_over_providing_needs_a_reason(neighbours):
o = subcontract_order()
pid = o["purchase_order_id"]
got = must(api("POST", f"/purchase-orders/{pid}/subcontract-provisions", ALICE, {"line_no": 1, "from_location_id": "WH/STOCK"}))
(p,) = neighbours.inventory.provisions.values()
assert p["body"]["move_ref"] == f"pur:{pid}:1:prov:1" and sorted(c["product_id"] for c in p["body"]["components"]) == ["STEEL-BAR", "WIDGET"]
assert {x["product_id"]: str(x["provided"]) for x in got["provisions"]}["STEEL-BAR"].startswith("240")
s, b = api("POST", f"/purchase-orders/{pid}/subcontract-provisions", ALICE, {"line_no": 1, "from_location_id": "WH/STOCK", "components": [{"product_id": "STEEL-BAR", "quantity": "1"}]})
assert s == 409 and "over_provided" in str(b)
must(api("POST", f"/purchase-orders/{pid}/subcontract-provisions", ALICE, {"line_no": 1, "from_location_id": "WH/STOCK", "reason": "scrap allowance",
"components": [{"product_id": "STEEL-BAR", "quantity": "1"}]}))
assert len(neighbours.inventory.provisions) == 2
def test_pur57_a_component_short_on_hand_is_refused_by_inventory_and_nothing_is_recorded(neighbours):
o = subcontract_order()
neighbours.inventory.short_components.add("WIDGET")
s, b = api("POST", f"/purchase-orders/{o['purchase_order_id']}/subcontract-provisions", ALICE, {"line_no": 1, "from_location_id": "WH/STOCK"})
assert s == 409 and "WIDGET" in str(b) and neighbours.inventory.provisions == {}
assert api("POST", f"/purchase-orders/{o['purchase_order_id']}/subcontract-provisions", BUYER, {"line_no": 1, "from_location_id": "WH/STOCK"})[0] == 403
def test_pur58_the_receipt_consumes_the_provided_components_and_carries_the_fee(neighbours):
o = subcontract_order()
s, r = receive(o, quantity="10")
assert s == 200
(m,) = neighbours.inventory.moves.values()
body = m["body"]
assert body["source_kind"] == "subcontract_receipt" and body["service_fee"] == "18.0000"
assert sorted((c["product_id"], c["quantity"]) for c in body["consume_provided"])[0][0] == "STEEL-BAR"
neighbours.inventory.moves.clear()
neighbours.inventory.short_components.add("STEEL-BAR")
s, r2 = receive(o, quantity="0.5", note="DN-2")
assert s == 200 and "STEEL-BAR" in str(r2) # the receipt stands, the hand-over lists inventory's reason