dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_floor.py
"""On the floor: INV-50 putaway rules and bin capacity, INV-51 pick batches, INV-52 SSCC packing; INV-49 ATP."""
from decimal import Decimal
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, SALES, api, must, refused
LTR_MEASURES = {"gross_weight": "1.05", "weight_unit_code": "KGM", "volume": "0.001", "volume_unit_code": "MTQ"}
PK24_MEASURES = {"gross_weight": "19.2", "weight_unit_code": "KGM", "volume": "0.02", "volume_unit_code": "MTQ"}
def floor():
world.catalogue()
fakes.add_product("COLA-075", base="LTR", units={"PK24": (18, 1, PK24_MEASURES), "C62": (3, 4), "LTR": (1, 1, LTR_MEASURES)}, warehouse="PK24", sales="C62")
fakes.PRODUCTS["COLA-075"]["category_code"] = "drinks"
world.warehouse(bins=("A/02", "A/03"), zones=("A", "C"), extra_bins=[("A/01", {"max_quantity_base": "108", "pick_sequence": 5}), ("C/01", {}), ("IN", {}),
("OUT-STAGE", {})])
must(api("POST", "/locations/settings", MANAGER, {"location_id": "WH1/A/02", "pick_sequence": 1}))
def rules():
for seq, applies, strategy, target in ((1, "product:COLA-075", "fixed", "WH1/A/01"), (2, "category:drinks", "next_with_room", "WH1/A"),
(3, "*", "addition", "WH1/C")):
must(api("POST", "/putaway-rules", MANAGER, {"warehouse_id": "WH1", "seq": seq, "applies_to": applies, "strategy": strategy, "target": target}))
def test_inv50_putaway_fills_the_fixed_bin_to_its_capacity_then_the_next_bin_and_adds_to_existing_stock():
floor()
rules()
world.receive("COLA-075", "5", "PK24", price="21.60", location="WH1/A/01") # A/01 holds 90 of 108
world.receive("SCREW-M4", "1", "BX100", price="2.00", location="WH1/A/03")
world.receive("COLA-075", "8", "PK24", price="21.60", location="WH1/IN")
world.receive("SCREW-M4", "1", "BX100", price="2.00", location="WH1/IN")
prop = must(api("GET", "/putaway/proposal", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "8", "unit_code": "PK24"}))
assert [(s["to_location_id"], s["quantity_base"]) for s in prop["splits"]] == [("WH1/A/01", "18"), ("WH1/A/02", "126")]
done = must(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "8", "unit_code": "PK24", "request_id": "p1"}))
assert [(p["to_location_id"], p["quantity_base"]) for p in done["putaways"]] == [("WH1/A/01", "18"), ("WH1/A/02", "126")]
assert must(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "8", "unit_code": "PK24", "request_id": "p1"}))["replay"] is True
screws = must(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "SCREW-M4", "quantity": "1", "unit_code": "BX100"}))
assert screws["putaways"][0]["to_location_id"] == "WH1/A/03" # the bin that already holds it
assert must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "280.80" # no value change: 13 packs x 21.60
def test_inv50_a_bin_over_its_capacity_is_refused_naming_it_and_a_chosen_bin_is_recorded_with_who():
floor()
rules()
world.receive("COLA-075", "8", "PK24", price="21.60", location="WH1/IN")
body = refused(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "8", "unit_code": "PK24", "to_location_id": "WH1/A/01"}),
409, "WH1/A/01 would exceed its quantity capacity: room for 108 LTR, 144 LTR put away")
assert body["detail"]["gate"] == "bin_capacity"
must(api("POST", "/locations/settings", MANAGER, {"location_id": "WH1/A/03", "max_volume": "0.1"}))
refused(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "8", "unit_code": "PK24", "to_location_id": "WH1/A/03"}),
409, "would exceed its volume capacity")
ok = must(api("POST", "/putaway", OPERATOR, {"warehouse_id": "WH1", "product_id": "COLA-075", "quantity": "2", "unit_code": "PK24", "to_location_id": "WH1/A/02"}))
assert "chosen by operator" in ok["note"].replace("stock_operator", "operator") and "instead of WH1/A/01" in ok["note"]
assert must(api("GET", "/documents/" + ok["putaways"][0]["document_id"], OPERATOR))["reason"].startswith("chosen by")
def deliveries(n=3):
world.catalogue()
world.warehouse(bins=("A/02",), extra_bins=[("A/01", {"pick_sequence": 5}), ("OUT-STAGE", {})])
must(api("POST", "/locations/settings", MANAGER, {"location_id": "WH1/A/02", "pick_sequence": 1}))
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01")
world.receive("DRILL-500", "10", "C62", price="30.00", location="WH1/A/02")
cola = {"product_id": "COLA-075", "unit_code": "C62", "unit_numerator": 3, "unit_denominator": 4, "quantity": "12"}
drill = {"product_id": "DRILL-500", "unit_code": "C62", "unit_numerator": 1, "unit_denominator": 1, "quantity": "3"}
ids = []
for i, line in enumerate((cola, drill, cola), start=1):
fakes.add_order(f"SO-{i}", {1: line})
ids.append(must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": f"SO-{i}", "warehouse_id": "WH1"}))["document_id"])
return ids
def test_inv51_one_pick_list_in_walking_order_moves_to_staging_and_each_delivery_still_validates_on_its_own():
ids = deliveries()
batch = must(api("POST", "/pick-batches", MANAGER, {"delivery_document_ids": ids}))
bid = batch["pick_batch_id"]
assert bid == "WH1/PB/00001"
again = refused(api("POST", "/pick-batches", MANAGER, {"delivery_document_ids": [ids[0]]}), 409, "already in pick batch WH1/PB/00001")
assert again is not None
lst = must(api("GET", "/pick-batches", OPERATOR, {"pick_batch_id": bid}))["lines"]
assert [(l["location_id"], l["delivery_document_id"]) for l in lst] == [("WH1/A/02", ids[1]), ("WH1/A/01", ids[0]), ("WH1/A/01", ids[2])] # walking order
refused(api("POST", f"/deliveries/{ids[0]}/cancel", OPERATOR, {"reason": "x"}), 409, "pick batch WH1/PB/00001")
done = must(api("POST", "/pick-batches/pick", OPERATOR, {"pick_batch_id": bid}))
assert done["to_location_id"] == "WH1/OUT-STAGE" and must(api("POST", "/pick-batches/pick", OPERATOR, {"pick_batch_id": bid}))["replay"] is True
st = must(api("GET", "/stock/COLA-075", OPERATOR, {"warehouse_id": "WH1"}))
assert st["on_hand"] == "36"
refused(api("POST", "/pick-batches/remove", MANAGER, {"pick_batch_id": bid, "delivery_document_id": ids[0]}), 409, "already in staging")
for i in ids:
v = must(api("POST", f"/deliveries/{i}/validate", OPERATOR, {}))
assert v["lifecycle"] == "done" and v["moves"][0]["from_location_id"] == "WH1/OUT-STAGE"
assert fakes.ORDERS["SO-2"]["lines"][1]["delivered"] == Decimal(3)
assert must(api("GET", "/stock/COLA-075", OPERATOR, {"warehouse_id": "WH1"}))["on_hand"] == "18"
def test_inv51_a_delivery_leaves_the_batch_whole_before_picking_and_may_then_be_cancelled():
ids = deliveries()
bid = must(api("POST", "/pick-batches", MANAGER, {"delivery_document_ids": ids}))["pick_batch_id"]
out = must(api("POST", "/pick-batches/remove", MANAGER, {"pick_batch_id": bid, "delivery_document_id": ids[0]}))
assert out["remaining"] == ids[1:]
assert must(api("POST", f"/deliveries/{ids[0]}/cancel", OPERATOR, {"reason": "customer changed mind"}))
assert [l["delivery_document_id"] for l in must(api("GET", "/pick-batches", OPERATOR, {"pick_batch_id": bid}))["lines"]] == [ids[1], ids[2]]
def gs1_check(first17: str) -> int:
weights = [3 if i % 2 == 0 else 1 for i in range(17)]
return (10 - sum(int(c) * w for c, w in zip(first17, weights)) % 10) % 10
def test_inv52_packages_get_an_sscc_with_a_check_digit_and_a_gross_weight_and_loose_items_block_validation():
world.catalogue()
fakes.add_product("COLA-075", base="LTR", units={"PK24": (18, 1, PK24_MEASURES), "C62": (3, 4)}, warehouse="PK24")
fakes.add_product("DRILL-500", base="C62", units={"C62": (1, 1, {"gross_weight": "2.5", "weight_unit_code": "KGM"})})
world.warehouse()
world.receive("COLA-075", "6", "PK24", price="21.60", location="WH1/A/01")
world.receive("DRILL-500", "2", "C62", price="30.00", location="WH1/A/02")
must(api("POST", "/settings", MANAGER, {"key": "delivery_steps", "value": "pick_pack_ship"}))
must(api("POST", "/settings", MANAGER, {"key": "gs1_company_prefix", "value": "5012345"}))
fakes.add_order("SO-P", {1: {"product_id": "COLA-075", "unit_code": "PK24", "unit_numerator": 18, "unit_denominator": 1, "quantity": "6"},
2: {"product_id": "DRILL-500", "unit_code": "C62", "unit_numerator": 1, "unit_denominator": 1, "quantity": "2"}})
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-P", "warehouse_id": "WH1"}))["document_id"]
must(api("POST", f"/deliveries/{doc}/pick", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": "6", "unit_code": "PK24"},
{"line_no": 2, "location_id": "WH1/A/02", "quantity": "2", "unit_code": "C62"}]}))
must(api("POST", "/package-types", MANAGER, {"package_type_code": "BOX-L", "tare_weight": "0.6", "weight_unit_code": "KGM", "length": "60", "width": "40",
"height": "40", "dimension_unit_code": "CMT"}))
must(api("POST", "/package-types", MANAGER, {"package_type_code": "PALLET-EU", "tare_weight": "25", "weight_unit_code": "KGM"}))
body = refused(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {}), 409, "is not in any package")
assert body["detail"]["gate"] == "all_packed"
refused(api("POST", f"/deliveries/{doc}/pack", OPERATOR, {"packages": [{"package_type_code": "PALLET-EU", "contents": [
{"product_id": "COLA-075", "quantity": "7", "unit_code": "PK24"}]}]}), 409, "never holds more than was picked")
packed = must(api("POST", f"/deliveries/{doc}/pack", OPERATOR, {"packages": [
{"package_type_code": "PALLET-EU", "contents": [{"product_id": "COLA-075", "quantity": "4", "unit_code": "PK24"}]},
{"package_type_code": "BOX-L", "contents": [{"product_id": "COLA-075", "quantity": "2", "unit_code": "PK24"}, {"product_id": "DRILL-500", "quantity": "2", "unit_code": "C62"}]}]}))
pallet, box = packed["packages"]
assert pallet["gross_weight"] == "101.8" and box["gross_weight"] == "44" and pallet["weight_unit_code"] == "KGM"
for p, n in ((pallet, 1), (box, 2)):
assert len(p["sscc"]) == 18 and p["sscc"].startswith("05012345") and p["sscc"][:-1].endswith(str(n).rjust(9, "0"))
assert int(p["sscc"][-1]) == gs1_check(p["sscc"][:-1])
ship = must(api("GET", f"/documents/{doc}/shipping", OPERATOR))
assert [p["sscc"] for p in ship["packages"]] == sorted([pallet["sscc"], box["sscc"]])
assert must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {}))["lifecycle"] == "done"
refused(api("POST", f"/deliveries/{doc}/pack", OPERATOR, {"packages": []}), 409, "is done")
fakes.add_order("SO-Q", {1: {"product_id": "DRILL-500", "unit_code": "C62", "unit_numerator": 1, "unit_denominator": 1, "quantity": "1"}})
world.receive("DRILL-500", "1", "C62", price="30.00", location="WH1/A/02")
d2 = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-Q", "warehouse_id": "WH1"}))["document_id"]
must(api("POST", f"/deliveries/{d2}/pick", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/02", "quantity": "1", "unit_code": "C62"}]}))
nxt = must(api("POST", f"/deliveries/{d2}/pack", OPERATOR, {"packages": [{"package_type_code": "BOX-L", "contents": [{"product_id": "DRILL-500", "quantity": "1", "unit_code": "C62"}]}]}))
assert nxt["packages"][0]["sscc"][:-1].endswith("000000003") # the series never reuses a number
def test_inv49_atp_counts_stock_now_and_purchase_and_transfer_receipts_from_the_day_they_are_usable():
world.catalogue()
world.warehouse()
world.receive("COLA-075", "5", "PK24", price="21.60", location="WH1/A/01") # 90 LTR
fakes.EXPECTED_RECEIPTS.append({"purchase_order_id": "PO-ATP", "line_no": 1, "product_id": "COLA-075", "warehouse_id": "WH1", "expected_on": "2026-11-25",
"quantity_base_outstanding": "36", "special_stock": "none"})
now = must(api("GET", "/atp/COLA-075", SALES, {"warehouse_id": "WH1", "quantity": "100", "unit_code": "LTR"}))
assert "2026-11-25" in str(now) and "90" in str(now)
early = must(api("GET", "/atp/COLA-075", SALES, {"warehouse_id": "WH1", "quantity": "60", "unit_code": "LTR", "requested_on": "2026-11-21"}))
assert "2026-11-21" in str(early)
assert api("GET", "/atp/COLA-075", person_without(), {"warehouse_id": "WH1", "quantity": "1", "unit_code": "LTR"})[0] == 404
def person_without():
from conftest import NOBODY
return NOBODY