dodil/erp-brick-sales-order
DODIL ERP brick: sales order
tests/test_slice_a.py
"""Slice A of sales-order v6: SO-01 (the accepted quote, priced by product), SO-23 (a stocked line is delivered only by inventory),
SO-25 (reservations on confirm, amend and cancel), SO-26 (exact base quantity), and the order-number claim."""
from decimal import Decimal as D
import pytest
from conftest import ADMIN, BILLER, FULFILLER, INVENTORY, REP, SYSTEM, api, confirm, deliver, get, hand, must, refused
from erp_platform.repository import CLAIM_TABLES, as_conflict
from sqlalchemy.exc import IntegrityError
PF = {"price_list_id": "LIST-GBP", "rule_seq": 1, "row": {"unit_code": "C62", "valid_from": "2026-10-01"}}
def ln(o, n):
return next(x for x in o["lines"] if x["line_no"] == n)
def quote(lines, version=2):
return {"quote_id": "Q-2026-0042", "quote_version": version, "bp_id": 7311, "currency": "GBP", "owner": "sales_rep@shop.example", "lines": lines}
def qline(n, product="COLA-075", qty="10", **kw):
return {"line_no": n, "product_id": product, "quantity": qty, "list_price": "1.20", "unit_price": "1.00", "unit_code": "C62", "priced_from": PF, **kw}
# ── SO-01 ────────────────────────────────────────────────────────────────────────────────────
def test_SO_01_a_quote_line_not_priced_by_product_is_refused_naming_the_first_and_nothing_is_made():
q = quote([qline(1), qline(2, "FEE", "1", priced_from=None), qline(3, "SETUP", "1", priced_from=None)])
said = refused(api("POST", "/sales-orders/from-quote", SYSTEM, q))
assert "priced_by_product" in said and "quote line 2 (FEE)" in said and "re-price the quote from product before it is sent" in said
assert must(api("GET", "/sales-orders", ADMIN))["orders"] == [] # no order, no tax call, no product call
q["lines"][1]["priced_from"] = {"rule_seq": 3} # a rule with no price list is not a product price either
assert api("POST", "/sales-orders/from-quote", SYSTEM, q)[0] == 422
def test_SO_01_a_retry_of_a_priced_quote_returns_the_same_order_even_if_the_book_has_changed(neighbours):
q = quote([qline(1)])
first = must(api("POST", "/sales-orders/from-quote", SYSTEM, q))
q["lines"][0]["priced_from"] = None # a retry whose body would now be refused still answers the order made
again = must(api("POST", "/sales-orders/from-quote", SYSTEM, q))
assert again["sales_order_id"] == first["sales_order_id"] and again["version"] == first["version"]
def test_SO_01_the_line_keeps_the_quotes_price_source_and_pricing_unit_and_nets_from_it():
o = must(api("POST", "/sales-orders/from-quote", SYSTEM, quote([qline(1, qty="250", unit_price="2.2050", pricing_quantity=100, priced_on="2026-11-10")])))
one = ln(o, 1)
assert one["priced_from"] == "quote:Q-2026-0042@v2#1 (LIST-GBP rule 1 on 2026-11-10)" and one["pricing_quantity"] == 100
assert D(one["unit_price"]) == D("2.2050") and D(one["net_amount"]) == D("5.51") # 250 x 2.2050 / 100 = 5.5125, rounded once; never a piece price
assert o["source_kind"] == "quote" and o["lifecycle"] == "draft"
def test_SO_01_a_price_per_100_is_invoiced_with_its_pricing_quantity_so_the_invoice_nets_the_same(neighbours):
o = confirm(must(api("POST", "/sales-orders/from-quote", SYSTEM, quote([qline(1, qty="250", unit_price="2.2050", pricing_quantity=100)]))))
must(api("POST", f"/sales-orders/{o['sales_order_id']}/request-invoice", BILLER, {}))
sent = neighbours.invoicing.drafts[f"so:{o['sales_order_id']}:1"]["lines"][0]
assert sent["pricing_quantity"] == 100 and sent["pricing_unit_code"] == "C62" and D(sent["net_amount"]) == D("5.51")
def test_SO_01_a_pricing_unit_the_product_no_longer_answers_is_refused_not_guessed():
said = refused(api("POST", "/sales-orders/from-quote", SYSTEM, quote([qline(1, pricing_unit_code="BX500", pricing_quantity=100)])), 422)
assert "re-price the quote from product" in said
# ── SO-23, SO-25, SO-26 ──────────────────────────────────────────────────────────────────────
def stocked_order(inventory, qty="48"):
o = hand(lines=[{"product_id": "COLA-075", "quantity": qty, "unit_code": "C62"}, {"product_id": "SETUP", "quantity": "3", "unit_code": "HUR"}])
return o, confirm(o)
def test_SO_25_confirm_reserves_at_revision_1_and_only_the_stocked_line_is_stocked(inventory):
o, c = stocked_order(inventory)
kind, oid, rev, lines = inventory.calls[0]
assert (kind, oid, rev) == ("reserve", o["sales_order_id"], 1) and [x["line_no"] for x in lines] == [1, 2]
assert (ln(c, 1)["stocked"], ln(c, 2)["stocked"]) == (True, False) # inventory ignored the service line: stocked stays false
assert D(ln(c, 1)["reserved_base"]) == 48 and D(ln(c, 1)["short_base"]) == 0
assert confirm(c) and len(inventory.calls) == 1 # confirming again calls nothing
def test_SO_25_a_short_reservation_is_shown_and_never_blocks_the_confirmation(inventory):
inventory.stock["COLA-075"] = D("10")
_, c = stocked_order(inventory)
assert c["lifecycle"] == "confirmed" and D(ln(c, 1)["reserved_base"]) == 10 and D(ln(c, 1)["short_base"]) == 38
def test_SO_25_inventory_that_does_not_answer_blocks_the_confirmation_and_the_order_stays_a_draft(inventory):
inventory.down = True
o = hand()
s, body = api("POST", f"/sales-orders/{o['sales_order_id']}/confirm", REP, {})
assert s == 502 and "inventory" in body["detail"]
assert get(o["sales_order_id"])["lifecycle"] == "draft"
inventory.down = False
assert confirm(o)["lifecycle"] == "confirmed" # the retry reserves once and confirms
def test_SO_25_an_amendment_reserves_again_at_the_next_revision_and_cancel_releases_with_the_reason(inventory):
o, c = stocked_order(inventory)
sid = o["sales_order_id"]
must(api("POST", f"/sales-orders/{sid}/amend", REP, {"reason": "they want more", "changes": [{"line_no": 1, "quantity": "60"}]}))
assert len(inventory.calls) == 1 # a pending amendment reserves nothing: it is not in force yet
a = must(api("POST", f"/sales-orders/{sid}/revisions/2/approve", ADMIN, {}))
assert a["revision"] == 2
kind, _, rev, lines = inventory.calls[-1]
assert (kind, rev) == ("reserve", 2) and next(x for x in lines if x["line_no"] == 1)["quantity"] == "60"
assert D(ln(a, 1)["reserved_base"]) == 60 and ln(a, 1)["stocked"] is True
must(api("POST", f"/sales-orders/{sid}/cancel", ADMIN, {"reason": "customer withdrew"}))
assert inventory.calls[-1] == ("release", sid, "customer withdrew")
n = len(inventory.calls)
must(api("POST", f"/sales-orders/{sid}/cancel", ADMIN, {"reason": "customer withdrew"}))
assert len(inventory.calls) == n # a repeat changes nothing, calls nothing
def test_SO_23_inventory_records_a_delivery_once_a_hand_delivery_on_a_stocked_line_is_refused(inventory):
o, _ = stocked_order(inventory)
must(deliver(o, 1, 24, "inv:WH1/OUT/00001", who=INVENTORY, source="inventory"))
again = must(deliver(o, 1, 24, "inv:WH1/OUT/00001", who=INVENTORY, source="inventory")) # however often it is retried
assert D(ln(again, 1)["delivered"]) == 24 and len([d for d in again["deliveries"] if d["line_no"] == 1]) == 1
assert D(again["deliveries"][0]["quantity_base"]) == 24 and again["deliveries"][0]["source"] == "inventory"
said = refused(deliver(o, 1, 5, "DN-1", who=FULFILLER))
assert "stocked_from_inventory" in said and "COLA-075 is stocked: its deliveries come from inventory" in said
must(deliver(o, 2, 3, "DN-2", who=FULFILLER)) # the 3 HUR by hand, as in v2
assert D(get(o["sales_order_id"])["lines"][0]["delivered"]) == 24
def test_SO_23_the_orders_rules_apply_unchanged_to_inventorys_call(inventory):
o, _ = stocked_order(inventory)
must(deliver(o, 1, 40, "inv:WH1/OUT/00001", who=INVENTORY, source="inventory"))
said = refused(deliver(o, 1, 9, "inv:WH1/OUT/00002", who=INVENTORY, source="inventory"))
assert "within_tolerance" in said and "would make 49 of 48 ordered" in said
must(deliver(o, 1, -4, "inv:WH1/IN/00001", who=INVENTORY, source="inventory", corrects_ref="inv:WH1/OUT/00001")) # a customer return
assert D(get(o["sales_order_id"])["lines"][0]["delivered"]) == 36
assert "stocked_from_inventory" in refused(deliver(o, 2, 1, "inv:WH1/OUT/00003", who=INVENTORY, source="inventory")) # a service line is never inventory's
def test_SO_23_only_inventorys_service_client_may_name_source_inventory(inventory):
o, _ = stocked_order(inventory)
s, body = deliver(o, 1, 1, "inv:WH1/OUT/00009", who=FULFILLER, source="inventory") # a fulfiller holds so:order:deliver, not inventory's identity
assert s == 403 and "service client" in body["detail"]
assert deliver(o, 1, 1, "x", who=FULFILLER, source="warehouse")[0] == 422
def test_SO_23_without_inventory_every_line_is_delivered_by_hand():
o = confirm(hand(lines=[{"product_id": "COLA-075", "quantity": "48", "unit_code": "C62"}]))
must(deliver(o, 1, 5, "DN-1", who=FULFILLER))
def test_SO_26_a_delivery_converts_exactly_to_the_base_unit_and_an_inexact_one_is_refused_unrecorded():
o = confirm(hand(lines=[{"product_id": "JUICE-075", "quantity": "30", "unit_code": "C62"}]))
got = must(deliver(o, 1, 24, "DN-1", who=FULFILLER))
assert D(got["deliveries"][0]["quantity_base"]) == D("18.000") and D(ln(got, 1)["delivered"]) == 24
said = refused(deliver(o, 1, "0.3333", "DN-2", who=FULFILLER))
assert "exact_base" in said and "0.3333 C62 is not an exact quantity of LTR to 3 decimals" in said
assert len(get(o["sales_order_id"])["deliveries"]) == 1 # never rounded, nothing recorded
ok = must(deliver(o, 1, "0.4", "DN-3", who=FULFILLER)) # 0.4 x 3/4 = 0.3: exact to 3 decimals
assert D(ok["deliveries"][-1]["quantity_base"]) == D("0.3")
# ── the order-number claim ────────────────────────────────────────────────────────────────────
def test_the_order_number_claim_is_declared_and_a_duplicate_is_a_conflict(engine):
from brick_sales_order.infrastructure.tables import so_order_no_claim
assert "so_order_no_claim" in CLAIM_TABLES
with pytest.raises(IntegrityError) as lost:
with engine.begin() as conn:
for oid in ("a", "b"):
conn.execute(so_order_no_claim.insert().values(order_no="SO-2026-0001", sales_order_id=oid))
assert as_conflict(lost.value) is not None # a commit-time duplicate is a 409, not a 500
def test_two_orders_racing_for_one_number_the_second_is_refused_never_silently_given_it(engine):
"""The check passes for both (each reads before the other writes); the INSERT is plain, so the second claim fails."""
from sqlalchemy.sql import Select
from brick_sales_order.infrastructure.tables import Repository
class Blind: # a connection whose reads see nothing: the other claimer has not committed yet
def __init__(self, conn):
self.conn = conn
def execute(self, stmt, *a, **kw):
if isinstance(stmt, Select):
return type("Nothing", (), {"first": staticmethod(lambda: None)})()
return self.conn.execute(stmt, *a, **kw)
with engine.begin() as conn:
assert Repository(conn).order_no_claims.claim("SO-2026-0001", "order-a") is True
with pytest.raises(IntegrityError) as lost:
with engine.begin() as conn:
Repository(Blind(conn)).order_no_claims.claim("SO-2026-0001", "order-b")
assert as_conflict(lost.value) is not None
def test_a_claim_lost_at_commit_is_a_409_never_a_500():
"""DataK3 can refuse the loser of a claim race at COMMIT, after the use case returned (live check, w2solab): the default repository answers
that as the retry-safe 409, and lets any other integrity failure through."""
from contextlib import contextmanager
from erp_platform.bricks import Refused
from brick_sales_order.application import use_cases
class Engine:
def __init__(self, text):
self.text = text
@contextmanager
def begin(self):
yield object()
raise IntegrityError("COMMIT", {}, Exception(self.text))
with pytest.raises(Refused) as lost:
with use_cases._on_engine(Engine("duplicate key value violates unique constraint (SQLSTATE 23505)")):
pass
assert lost.value.status == 409
with pytest.raises(IntegrityError):
with use_cases._on_engine(Engine("null value in column violates not-null constraint")):
pass