dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_doors.py
"""Both doors answer alike: the same use case as an API route and as an MCP tool; replays; the process view; settings."""
from decimal import Decimal
from conftest import ADMIN, ALICE, FULFILLER, OWNER, REP, SYSTEM, api, confirmed, deliver, hand, must, tool
D = Decimal
W = [{"product_id": "WIDGET", "quantity": "12", "unit_code": "C62"}]
def test_a_draft_through_both_doors_reads_the_same():
made = tool(REP, "draft_by_hand", {"bp_id": "7311", "currency": "GBP", "lines": W})
assert made[0] == "ok" and made[1]["lifecycle"] == "draft" and made[1]["bp_id"] == "7311"
sid = made[1]["sales_order_id"]
via_api = must(api("GET", f"/sales-orders/{sid}", REP))
assert tool(REP, "get_sales_order", {"sales_order_id": sid}) == ("ok", via_api)
assert tool(REP, "get_order_readiness", {"sales_order_id": sid})[1]["readiness"]["confirm"][0]["gate"] == "customer_active"
assert tool(REP, "confirm_order", {"sales_order_id": sid})[1]["order_no"] == "SO-2026-0001"
assert tool(FULFILLER, "record_delivery", {"sales_order_id": sid, "line_no": 1, "quantity": "2", "delivery_ref": "D1"})[0] == "ok"
def test_a_refusal_through_the_tool_says_the_same_words():
status, body = api("POST", f"/sales-orders/{hand()['sales_order_id']}/cancel", REP, {"reason": "x"})
assert status == 403
o = confirmed(lines=W)
deliver(o, 1, 1, "D1")
a = api("POST", f"/sales-orders/{o['sales_order_id']}/cancel", ADMIN, {"reason": "x"})
t = tool(ADMIN, "cancel_order", {"sales_order_id": o["sales_order_id"], "reason": "x"})
assert a[0] == 409 and t[0] == "refused" and t[1].endswith(a[1]["detail"]) and "nothing_delivered" in t[1]
def test_idempotent_commands_answer_a_replay_the_same_through_both_doors():
o = hand()
sid = o["sales_order_id"]
for call in (lambda: api("POST", f"/sales-orders/{sid}/confirm", REP, {}), lambda: tool(REP, "confirm_order", {"sales_order_id": sid})):
first = call()
assert call() == first
c = confirmed(lines=W)
d1 = deliver(c, 1, 3, "R1")
assert deliver(c, 1, 3, "R1") == d1
p = hand()
first = api("POST", f"/sales-orders/{p['sales_order_id']}/cancel-draft", REP, {})
assert api("POST", f"/sales-orders/{p['sales_order_id']}/cancel-draft", REP, {}) == first
assert tool(REP, "cancel_draft_order", {"sales_order_id": p["sales_order_id"]})[1] == first[1]
def test_a_command_that_makes_something_new_takes_an_idempotency_key():
keyed = {**REP, "Idempotency-Key": "k-1"}
body = {"bp_id": 7311, "currency": "GBP", "lines": W}
a, b = api("POST", "/sales-orders", keyed, body), api("POST", "/sales-orders", keyed, body)
assert a == b and len(must(api("GET", "/sales-orders", ALICE))["orders"]) == 1
assert len(must(api("GET", "/sales-orders", ALICE, {}))["orders"]) == 1
api("POST", "/sales-orders", REP, body)
assert len(must(api("GET", "/sales-orders", ALICE))["orders"]) == 2
def test_a_stale_version_is_a_409_and_nothing_is_written():
o = hand()
sid = o["sales_order_id"]
must(api("POST", f"/sales-orders/{sid}/customer-po", REP, {"customer_po_ref": "PO-1"}))
s, body = api("POST", f"/sales-orders/{sid}/customer-po", REP, {"customer_po_ref": "PO-2", "version": o["version"]})
assert s == 409 and "changed since you opened it" in body["detail"]
assert must(api("GET", f"/sales-orders/{sid}", REP))["customer_po_ref"] == "PO-1"
def test_the_process_view_shows_the_workflow_as_bpmn():
status, body = api("GET", "/process/bpmn", ALICE)
assert status == 200 and body["processes"] == ["sales_order", "return_order", "rebate_agreement"]
def test_the_settings_are_seeded_dated_and_typed():
body = must(api("GET", "/settings", OWNER))
kinds = {s["setting_key"]: (s["kind"], s["value"]) for s in body["settings"]}
assert kinds == {"order_number_format": ("str", "SO-{yyyy}-{seq:4}"), "number_restarts_yearly": ("bool", True),
"confirm_on_acceptance": ("bool", False), "amendment_needs_approval": ("bool", True), "default_deposit_pct": ("int", 0),
"lock_done_orders": ("bool", True), "business_timezone": ("str", "Europe/London"), "contract_creates_order": ("bool", False),
"ai.ai_order_from_po.mode": ("mode", "off"), "ai.ai_check_po.mode": ("mode", "off"),
"delivery_multiple_mode": ("mode", "warn"), "rebate_accrual_mode": ("str", ""), "rebate_accounts": ("str", ""),
"company_country": ("str", "GB"), "default_warehouse_id": ("str", ""), "pick_pack_days": ("int", 1),
"reschedule_mode": ("mode", "later_only"), "return_credit_on": ("mode", "inspection"), "free_of_charge_limit": ("str", "250.00"),
"default_deposit_deduction_mode": ("mode", "proportional")}
assert body["business_timezone"] == "Europe/London" and body["today"] == "2026-11-20"
assert api("POST", "/settings/order_number_format", ADMIN, {"value": 7, "valid_from": "2026-11-21"})[0] == 422
assert api("POST", "/settings/no_such", ADMIN, {"value": 7, "valid_from": "2026-11-21"})[0] == 404
assert api("POST", "/controls/ai.ai_check_po.mode", OWNER, {"value": "auto", "valid_from": "2026-11-21"})[0] == 422
def test_the_number_series_is_state_the_machine_may_move_up_never_down():
confirmed(lines=W)
assert must(api("GET", "/number-series/SO-2026", ALICE))["high_water_mark"] == 1
assert must(api("POST", "/number-series/SO-2026", SYSTEM, {"high_water_mark": 40}))["high_water_mark"] == 40
assert api("POST", "/number-series/SO-2026", SYSTEM, {"high_water_mark": 40})[0] == 200
assert api("POST", "/number-series/SO-2026", SYSTEM, {"high_water_mark": 10})[0] == 409
assert confirmed(lines=W)["order_no"] == "SO-2026-0041"