dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_deliveries.py
"""Delivering: INV-07 pick and ship, INV-08 the last unit, INV-13 reserve by hand, INV-22 reservations, INV-23 the order hears of it."""
from decimal import Decimal
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, SYSTEM, api, must, on_day, refused, tool
BOTTLES = {"product_id": "COLA-075", "unit_code": "C62", "quantity": "48", "unit_numerator": 3, "unit_denominator": 4}
def stocked(pallets="2", order_qty="48", reserve=True, order="SO-2026-0012"):
world.catalogue()
world.warehouse()
world.receive("COLA-075", pallets, "PK24", price="21.60", location="WH1/A/01")
fakes.add_order(order, {1: {**BOTTLES, "quantity": order_qty}})
if reserve:
must(api("POST", f"/reservations/sales-orders/{order}", SYSTEM, {"revision": 1, "warehouse_id": "WH1",
"lines": [{"line_no": 1, **{**BOTTLES, "quantity": order_qty}}]}))
def test_inv22_the_order_reserves_and_a_repeat_changes_nothing():
world.catalogue()
world.warehouse()
world.receive("COLA-075", "60", "C62", price="0.90", location="WH1/A/01") # 45 LTR on hand
body = {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}
got = must(api("POST", "/reservations/sales-orders/SO-2026-0012", SYSTEM, body))
line = got["lines"][0]
assert (line["reserved_base"], line["short_base"], line["base_unit_code"]) == ("36", "0", "LTR")
assert line["readiness"][0]["gate"] == "stock_reserved" and line["readiness"][0]["met"] is True
again = must(api("POST", "/reservations/sales-orders/SO-2026-0012", SYSTEM, body))
assert again["replay"] is True and again["lines"][0]["reserved_base"] == "36"
av = must(api("GET", "/availability/COLA-075", SALES_REP()))
assert av["warehouses"] == [{"warehouse_id": "WH1", "on_hand": "45", "reserved": "36", "available": "9", "in_transit_in": "0"}]
rev2 = must(api("POST", "/reservations/sales-orders/SO-2026-0012", SYSTEM, {**body, "revision": 2, "lines": [{"line_no": 1, **{**BOTTLES, "quantity": "24"}}]}))
assert rev2["lines"][0]["reserved_base"] == "18" # revision 2 reduces the reservation
freed = must(api("POST", "/reservations/sales-orders/SO-2026-0012/release", SYSTEM, {"reason": "cancelled"}))
assert freed["released_base_by_line"] == [{"line_no": 1, "quantity_base": "18"}]
assert must(api("POST", "/reservations/sales-orders/SO-2026-0012/release", SYSTEM, {}))["released_base_by_line"] == []
assert must(api("GET", "/availability/COLA-075", SALES_REP()))["warehouses"][0]["available"] == "45"
def SALES_REP():
from conftest import SALES
return SALES
def test_inv22_a_service_line_is_ignored_and_said_so():
stocked(reserve=False)
got = must(api("POST", "/reservations/sales-orders/SO-9", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [
{"line_no": 1, "product_id": "HOST-M", "unit_code": "MON", "quantity": "1"}]}))
assert "service" in got["lines"][0]["ignored"]
def test_inv22_short_when_stock_is_not_there_and_a_machine_grant_is_needed():
stocked(pallets="1", reserve=False)
got = must(api("POST", "/reservations/sales-orders/SO-2026-0012", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}))
line = got["lines"][0]
assert (line["reserved_base"], line["short_base"]) == ("18", "18") and line["readiness"][0]["met"] is False
assert api("POST", "/reservations/sales-orders/SO-1", MANAGER, {"revision": 1, "lines": []})[0] == 403
def test_inv07_pick_a_part_and_the_order_hears_of_it_once():
stocked()
made = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"}))
doc = made["document_id"]
assert doc == "dlv:so:SO-2026-0012:1" and made["lines"][0]["quantity"] == "48" and made["lines"][0]["reserved_base"] == "36"
done = must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "quantity": "24", "unit_code": "C62"}]}))
assert "value_amount" not in str(done) # the floor never sees a value
assert done["number"] == "WH1/OUT/00001" and done["delivery_ref"] == "inv:WH1/OUT/00001" and done["lifecycle"] == "done"
(mv,) = must(api("GET", f"/documents/{doc}", ACCOUNTANT))["moves"]
assert (mv["from_location_id"], mv["to_location_id"], mv["quantity_base"], mv["value_amount"]) == ("WH1/A/01", "UK1:CUSTOMER", "18", "-21.60")
assert fakes.ORDERS["SO-2026-0012"]["lines"][1]["delivered"] == Decimal(24)
assert ("SO-2026-0012", "inv:WH1/OUT/00001", 1) in fakes.DELIVERY_REFS
assert done["back_order"]["lines"] == [{"line_no": 1, "quantity": "24", "unit_code": "C62"}]
got = must(api("GET", f"/deliveries/{done['back_order']['document_id']}", OPERATOR))
assert got["lifecycle"] == "ready" and got["back_order_of"] == doc and got["lines"][0]["reserved_base"] == "18" # the reservation dropped by 18
again = must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {}))
assert again["replay"] is True and fakes.ORDERS["SO-2026-0012"]["lines"][1]["delivered"] == Decimal(24)
assert must(api("GET", "/stock/COLA-075", OPERATOR))["on_hand"] == "18"
def ACCOUNTANT_OPERATOR():
return OPERATOR
def test_inv07_picking_from_a_bin_that_does_not_hold_it_names_what_the_bin_holds():
stocked()
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"}))["document_id"]
refused(api("POST", f"/deliveries/{doc}/pick", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/02", "quantity": "1", "unit_code": "C62"}]}),
409, "WH1/A/02 holds 0 LTR of COLA-075")
refused(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {}), 409, "nothing is picked yet")
def test_inv23_a_refusal_by_the_order_refuses_the_whole_validation_and_writes_nothing():
stocked(order_qty="10")
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"}))["document_id"]
body = refused(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "quantity": "11", "unit_code": "C62"}]}), 409, "more than the 10 ordered")
assert body["detail"]["gate"] == "order_accepts"
assert must(api("GET", "/stock/COLA-075", OPERATOR))["on_hand"] == "36"
assert must(api("GET", f"/deliveries/{doc}", OPERATOR))["lifecycle"] == "ready"
fakes.ORDERS["SO-2026-0012"]["lifecycle"] = "cancelled"
assert api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": "1", "unit_code": "C62"}]})[0] == 409
def test_inv08_two_pickers_one_last_unit_exactly_one_wins_and_the_loser_writes_and_tells_nothing():
world.catalogue()
world.warehouse()
world.receive("DRILL-500", "1", "C62", price="80.00", location="WH1/A/03")
for so in ("SO-1", "SO-2"):
fakes.add_order(so, {1: {"product_id": "DRILL-500", "unit_code": "C62", "quantity": "1"}})
d1 = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-1", "warehouse_id": "WH1"}))["document_id"]
d2 = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2", "warehouse_id": "WH1"}))["document_id"]
pick = {"picks": [{"line_no": 1, "location_id": "WH1/A/03", "quantity": "1", "unit_code": "C62"}]}
# both read the same snapshot: the second one's claim read comes back empty, as in a transaction that began earlier
from unittest import mock
from brick_inventory.infrastructure.tables import Repository
must(api("POST", f"/deliveries/{d1}/validate", OPERATOR, pick))
from brick_inventory.application.unit import Unit
with mock.patch.object(Repository, "last_claim_no", lambda self, key: 0), \
mock.patch.object(Unit, "stock", lambda self, key: Decimal(1)):
body = refused(api("POST", f"/deliveries/{d2}/validate", OPERATOR, pick), 409, "WH1/A/03 holds 0")
assert "another delivery took it" in str(body)
assert ("SO-2", "inv:WH1/OUT/00002", 1) not in fakes.DELIVERY_REFS and not any(k[0] == "SO-2" for k in fakes.DELIVERY_REFS)
assert fakes.ORDERS["SO-2"]["lines"][1]["delivered"] == 0
assert must(api("GET", "/stock/DRILL-500", OPERATOR))["on_hand"] == "0" # on hand never shows -1
assert must(api("GET", f"/deliveries/{d2}", OPERATOR))["lifecycle"] == "ready"
def test_inv13_the_manager_moves_a_reservation_from_one_order_to_another():
world.catalogue()
world.warehouse()
world.receive("COLA-075", "36", "C62", price="0.90", location="WH1/A/01") # 27 LTR
fakes.add_order("SO-12", {1: {**BOTTLES, "quantity": "36"}})
fakes.add_order("SO-14", {1: {**BOTTLES, "quantity": "48"}})
must(api("POST", "/reservations/sales-orders/SO-12", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **{**BOTTLES, "quantity": "36"}}]}))
must(api("POST", "/reservations/sales-orders/SO-14", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}))
refused(api("POST", "/reservations/hand", MANAGER, {"sales_order_id": "SO-14", "line_no": 1, "quantity_base": "9", "reason": "VIP"}), 409, "only 0 available")
must(api("POST", "/reservations/hand", MANAGER, {"sales_order_id": "SO-12", "line_no": 1, "quantity_base": "-9", "reason": "VIP order first"}))
got = must(api("POST", "/reservations/hand", MANAGER, {"sales_order_id": "SO-14", "line_no": 1, "quantity_base": "9", "reason": "VIP order first"}))
assert got["held_base"] == "9" and got["reason"] == "VIP order first"
refused(api("POST", "/reservations/hand", OPERATOR, {"sales_order_id": "SO-14", "line_no": 1, "quantity_base": "1", "reason": "x"}), 403, "inv:reservation:write")
def test_inv35_a_delivery_shows_its_weight_copied_when_written_and_can_require_it():
world.catalogue()
fakes.add_product("COLA-075", base="LTR", units={
"PK24": (18, 1, {"gross_weight": "19.2", "net_weight": "18.0", "weight_unit_code": "KGM", "volume": "0.027", "volume_unit_code": "MTQ"}),
"C62": (3, 4, {"gross_weight": "0.79", "net_weight": "0.75", "weight_unit_code": "KGM"})}, warehouse="PK24")
world.warehouse()
world.receive("COLA-075", "4", "PK24", location="WH1/A/01")
fakes.add_order("SO-35", {1: {"product_id": "COLA-075", "unit_code": "PK24", "quantity": "3", "unit_numerator": 18, "unit_denominator": 1},
2: {"product_id": "COLA-075", "unit_code": "C62", "quantity": "6", "unit_numerator": 3, "unit_denominator": 4}})
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-35", "warehouse_id": "WH1"}))["document_id"]
must(api("POST", f"/deliveries/{doc}/pick", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "quantity": "3", "unit_code": "PK24"},
{"line_no": 2, "location_id": "WH1/A/01", "quantity": "6", "unit_code": "C62"}]}))
must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {}))
ship = must(api("GET", f"/documents/{doc}/shipping", OPERATOR))
assert (ship["gross_weight"], ship["net_weight"], ship["volume"]) == ("62.34", "58.5", "0.081")
assert ship["volume_unknown"] == "volume unknown for 1 of 2 lines"
# a measure changed on the product later never changes a document already written
fakes.PRODUCTS["COLA-075"]["units"][0]["measures"]["gross_weight"] = "99"
assert must(api("GET", f"/documents/{doc}/shipping", OPERATOR))["gross_weight"] == "62.34"
def test_inv35_with_delivery_needs_weights_a_line_of_unknown_weight_refuses_validation():
stocked()
must(api("POST", "/settings", MANAGER, {"key": "delivery_needs_weights", "value": "true"}))
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"}))["document_id"]
refused(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": "1", "unit_code": "C62"}]}),
409, "has no weight")
def test_sc_inv07_today_at_midnight_is_the_business_date_not_the_utc_date():
stocked(order_qty="4")
with on_day("2026-10-31T23:30:00+00:00"): # 00:30 on 2026-11-01 in Europe/London? no: still BST->GMT on Oct 25; London is UTC+0
pass
from brick_inventory.infrastructure import business_time
business_time.SOURCE = type("S", (), {"rows": lambda self: [(business_time.EPOCH, "Pacific/Auckland")]})()
with on_day("2026-11-20T23:30:00+00:00"): # 12:30 on 2026-11-21 in Auckland
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"}))["document_id"]
must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": "4", "unit_code": "C62"}]}))
d = must(api("GET", f"/documents/{doc}", OPERATOR))
assert d["done_on"] == "2026-11-21" and d["moves"][0]["moved_on"] == "2026-11-21"
assert fakes.DELIVERY_REFS[("SO-2026-0012", "inv:WH1/OUT/00001", 1)]["delivered_on"] == "2026-11-21"
def test_deliveries_are_tools_for_agents_too():
stocked()
status, got = tool(OPERATOR, "create_delivery", {"sales_order_id": "SO-2026-0012", "warehouse_id": "WH1"})
assert status == "ok" and got["document_id"].startswith("dlv:so:")
def test_inv22_sales_orders_v5_body_priority_and_a_per_line_warehouse_is_taken():
"""SO-36 (aligned with INV-22): sales-order sends priority as normal | rush and, on a promised line, the warehouse with the
schedule. Both were refused 422 (priority as an int, warehouse_id an extra field), so no order with inventory installed confirmed."""
world.catalogue()
world.warehouse()
world.receive("COLA-075", "60", "C62", price="0.90", location="WH1/A/01")
body = {"revision": 1, "schedule_version": 1, "priority": "rush", "lines": [
{"line_no": 1, **BOTTLES, "warehouse_id": "WH1", "schedule": [{"confirmed_on": "2026-11-21", "quantity_base": "36"}]}]}
got = must(api("POST", "/reservations/sales-orders/SO-2026-0012", SYSTEM, body))
assert got["lines"][0]["reserved_base"] == "36"
assert must(api("POST", "/reservations/sales-orders/SO-2026-0013", SYSTEM, {"revision": 1, "priority": "normal", "lines": [
{"line_no": 1, **{**BOTTLES, "quantity": "4"}}]}))["lines"][0]["reserved_base"] == "3"
refused(api("POST", "/reservations/sales-orders/SO-9", SYSTEM, {"revision": 1, "priority": "soon", "lines": []}), 422)
refused(api("POST", "/reservations/sales-orders/SO-9", SYSTEM, {"revision": 1, "lines": [
{"line_no": 1, **BOTTLES, "warehouse_id": "WH1"}, {"line_no": 2, **BOTTLES, "warehouse_id": "WH2"}]}), 422, "more than one warehouse")
def test_inv07_an_order_that_takes_no_part_shipment_waits_until_it_is_reserved_in_full():
"""SO-37: partial_delivery is allowed | complete_line | complete_order. The fake answered True and the code tested the value
for truth, so a real complete_order (a non-empty string) shipped in parts. Found comparing the fake with the real order."""
world.catalogue()
world.warehouse()
world.receive("COLA-075", "1", "PK24", price="21.60", location="WH1/A/01") # 18 LTR: short of the 36 ordered
for order, mode, want in (("SO-W", "complete_order", "waiting"), ("SO-R", "allowed", "ready")):
fakes.add_order(order, {1: dict(BOTTLES)}, partial_delivery=mode)
must(api("POST", f"/reservations/sales-orders/{order}", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}))
made = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": order, "warehouse_id": "WH1"}))
assert made["lifecycle"] == want, (mode, made)
must(api("POST", f"/reservations/sales-orders/{order}/release", SYSTEM, {"reason": "next case"}))