dodil/erp-brick-purchasing
DODIL ERP brick: purchasing
tests/test_stories_agreements.py
"""Contracts — PUR-41 (approve once), PUR-42 (order against it, never past it)."""
from conftest import ALICE, APPROVER, BUYER, CLERK, SUPPLIER_A, SUPPLIER_B, api, must, on_day
LAPTOP = {"product_id": "LAPTOP", "unit_code": "C62"}
def value_contract(ref="C-D-1", bp=SUPPLIER_A):
return must(api("POST", "/agreements", BUYER, {"bp_id": bp, "kind": "value_contract", "supplier_ref": ref, "valid_from": "2026-11-20", "valid_to": "2027-12-31",
"target_value": "60000.00", "lines": [{**LAPTOP, "unit_price": "100.0000"}]}))
def quantity_contract(ref="C-E-1", bp=SUPPLIER_B):
return must(api("POST", "/agreements", BUYER, {"bp_id": bp, "kind": "quantity_contract", "supplier_ref": ref, "valid_from": "2026-11-20", "valid_to": "2027-12-31",
"lines": [{**LAPTOP, "target_quantity": "5000", "unit_price": "10.0000"}]}))
def approved(a):
must(api("POST", f"/agreements/{a['agreement_id']}/submit", BUYER, {}))
return must(api("POST", f"/agreements/{a['agreement_id']}/approve", APPROVER, {}))
def release(a, qty, line_no=1):
return must(api("POST", f"/agreements/{a['agreement_id']}/release-orders", BUYER, {"lines": [{"line_no": line_no, "quantity": qty}]}))
def confirm(o):
return api("POST", f"/purchase-orders/{o['purchase_order_id']}/confirm", BUYER, {})
def test_pur41_a_contract_is_keyed_numbered_approved_once_and_not_by_its_buyer():
from conftest import world
world()
a = value_contract()
assert a["agreement_id"] == "ag:7001:value_contract:2026-11-20:C-D-1" and a["status"] == "draft" and str(a["target"]) == "60000.00"
assert value_contract()["agreement_id"] == a["agreement_id"]
mine = must(api("POST", "/agreements", ALICE, {"bp_id": SUPPLIER_A, "kind": "value_contract", "supplier_ref": "ALICE", "valid_from": "2026-11-20",
"valid_to": "2027-12-31", "target_value": "100.00", "lines": [{**LAPTOP, "unit_price": "1.0000"}]}))
must(api("POST", f"/agreements/{mine['agreement_id']}/submit", ALICE, {}))
s, b = api("POST", f"/agreements/{mine['agreement_id']}/approve", ALICE, {})
assert s == 409 and "second_person" in str(b)
must(api("POST", f"/agreements/{a['agreement_id']}/submit", BUYER, {}))
assert api("POST", f"/agreements/{a['agreement_id']}/approve", CLERK, {})[0] == 403
ok = must(api("POST", f"/agreements/{a['agreement_id']}/approve", APPROVER, {}))
assert ok["status"] == "active" and ok["agreement_no"] == "AG-2026-0001" and ok["approved_by_email"] == "approver@shop.example"
def test_pur41_a_release_order_is_priced_from_the_agreement_and_needs_no_further_approval():
from conftest import world
world(threshold="0")
a = approved(value_contract())
o = release(a, "10")
assert str(o["lines"][0]["unit_price"]) == "100.0000" and o["lines"][0]["price_source"] == "agreement" and o["agreement_id"] == a["agreement_id"]
s, c = confirm(o)
assert s == 200 and c["status"] == "confirmed"
def test_pur41_a_revision_goes_back_to_approval_and_releases_continue_under_the_approved_one():
from conftest import world
world(threshold="0")
a = approved(value_contract())
must(api("POST", f"/agreements/{a['agreement_id']}/revise", BUYER, {"target_value": "90000.00"}))
assert confirm(release(a, "10"))[0] == 200
got = must(api("GET", f"/agreements/{a['agreement_id']}", BUYER))
assert str(got["target_value"]) == "60000.00" and got["revision"] == 1
must(api("POST", f"/agreements/{a['agreement_id']}/submit", BUYER, {}))
new = must(api("POST", f"/agreements/{a['agreement_id']}/approve", APPROVER, {}))
assert str(new["target_value"]) == "90000.00" and new["revision"] == 2
def test_pur42_a_value_release_past_what_is_left_is_refused_with_the_amount_left():
from conftest import world
world(threshold="0")
a = approved(value_contract())
assert confirm(release(a, "524"))[0] == 200 # 52,400.00 released
s, b = confirm(release(a, "90")) # 9,000.00 > 7,600.00 left
assert s == 409 and "within_agreement" in str(b) and "7,600.00 is left" in str(b)
got = must(api("GET", f"/agreements/{a['agreement_id']}", BUYER))
assert str(got["released_value"]) == "52400.00" and str(got["remaining_value"]) == "7600.00"
assert confirm(release(a, "76"))[0] == 200 # exactly what is left
def test_pur42_a_quantity_release_never_passes_the_target_and_after_valid_to_is_refused():
from conftest import world
world(threshold="0")
a = approved(quantity_contract())
assert confirm(release(a, "4800"))[0] == 200
assert confirm(release(a, "150"))[0] == 200 # 4,950 released
s, b = confirm(release(a, "150"))
assert s == 409 and "within_agreement" in str(b) and "50" in str(b)
got = must(api("GET", f"/agreements/{a['agreement_id']}", BUYER))
assert str(got["lines"][0]["remaining_base"]).startswith("50")
with on_day("2028-01-02T09:00:00+00:00"):
s, b = api("POST", f"/agreements/{a['agreement_id']}/release-orders", BUYER, {"lines": [{"line_no": 1, "quantity": "1"}]})
assert s == 409 and "within_agreement" in str(b)
def test_pur42_two_confirmations_on_one_snapshot_collide_on_the_agreements_claim_sequence():
import pytest
from conftest import world
from erp_platform.repository import Conflict
from test_stories_p2p import base_engine
from brick_purchasing.infrastructure.tables import Claims
world(threshold="0")
a = approved(quantity_contract())
with base_engine() as e:
c1, c2 = e.connect(), e.connect()
t1, t2 = c1.begin(), c2.begin()
scope = f"agreement:{a['agreement_id']}"
n1, n2 = Claims(c1).next_seq(scope), Claims(c2).next_seq(scope)
Claims(c1).claim(scope, n1)
t1.commit()
with pytest.raises(Conflict):
Claims(c2).claim(scope, n2)
t2.rollback()
def scheduling(ref="S-F-1", bp=SUPPLIER_A):
return must(api("POST", "/agreements", BUYER, {"bp_id": bp, "kind": "scheduling_agreement", "supplier_ref": ref, "valid_from": "2026-11-20", "valid_to": "2027-10-31",
"firm_zone_days": 7, "lines": [{**LAPTOP, "target_quantity": "100000", "unit_price": "0.5000"}]}))
def test_pur43_a_jit_release_opens_a_numbered_call_off_and_a_forecast_commits_nothing():
from conftest import world
world(threshold="0")
a = approved(scheduling())
aid = a["agreement_id"]
sch = must(api("POST", f"/agreements/{aid}/schedule", BUYER, {"lines": [{"line_no": 1, "delivery_date": d, "quantity": q} for d, q in
(("2026-11-23", "20000"), ("2026-11-30", "20000"), ("2026-12-14", "20000"))]}))
assert len(sch["schedule"]) == 3
fc = must(api("POST", f"/agreements/{aid}/releases", BUYER, {"kind": "forecast", "to": "f@x.example"}))
assert fc["release_no"] == 1 and fc["call_off_order_id"] == ""
assert not must(api("GET", "/purchase-orders", BUYER, {"status": "confirmed"}))["orders"]
jit = must(api("POST", f"/agreements/{aid}/releases", BUYER, {"kind": "jit", "to": "f@x.example", "through": "2026-11-30"}))
assert jit["release_no"] == 2 and jit["call_off_order_id"] == f"sa:{aid}:2"
o = must(api("GET", f"/purchase-orders/{jit['call_off_order_id']}", BUYER))
assert o["status"] == "confirmed" and str(o["lines"][0]["quantity"]).startswith("40000") and o["agreement_id"] == aid
def test_pur43_a_jit_line_inside_the_firm_zone_needs_a_reason_and_a_past_line_cannot_change():
from conftest import world
world(threshold="0")
aid = approved(scheduling())["agreement_id"]
must(api("POST", f"/agreements/{aid}/schedule", BUYER, {"lines": [{"line_no": 1, "delivery_date": "2026-11-23", "quantity": "20000"}, {"line_no": 1, "delivery_date": "2026-12-14", "quantity": "20000"}]}))
must(api("POST", f"/agreements/{aid}/releases", BUYER, {"kind": "jit", "to": "f@x.example", "through": "2026-11-23"}))
keep = {"line_no": 1, "delivery_date": "2026-12-14", "quantity": "20000"}
s, b = api("POST", f"/agreements/{aid}/schedule", BUYER, {"lines": [{"line_no": 1, "delivery_date": "2026-11-23", "quantity": "26000"}, keep]})
assert s == 409 and "firm zone" in str(b)
ok = must(api("POST", f"/agreements/{aid}/schedule", BUYER, {"lines": [{"line_no": 1, "delivery_date": "2026-11-23", "quantity": "26000", "reason": "supplier agreed by phone"}, keep]}))
assert str(ok["schedule"][0]["quantity"]).startswith("26000")
assert api("POST", f"/agreements/{aid}/schedule", BUYER, {"lines": [{"line_no": 1, "delivery_date": "2026-11-10", "quantity": "5"}, keep]})[0] == 422