dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_lots.py
"""Lots, serials and picking by expiry: INV-39 (FEFO / FIFO, minimum shelf life), INV-40 serials, INV-41 lot trace and block."""
import fakes
import world
from conftest import MANAGER, OPERATOR, OWNER, api, must, on_day, refused
def fefo_world(strategy="fefo"):
world.catalogue()
world.warehouse()
must(api("POST", "/settings", OWNER, {"key": "lots_enabled", "value": "true"}))
must(api("POST", "/stocked-products", MANAGER, {"product_id": "COLA-075", "tracking": "lot", "removal_strategy": strategy}))
world.receive("COLA-075", "1", "PK24", price="21.60", location="WH1/A/01", lot_code="L-2611", expires_on="2026-12-01")
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/02", lot_code="L-2612", expires_on="2027-02-01")
world.receive("COLA-075", "12", "C62", price="0.90", location="WH1/A/03", lot_code="L-2601", expires_on="2026-10-15")
fakes.add_order("SO-7311", {1: {"product_id": "COLA-075", "unit_code": "C62", "quantity": "24", "unit_numerator": 3, "unit_denominator": 4}},
min_remaining_shelf_life_days=60)
def test_inv39_fefo_proposes_the_lot_with_enough_shelf_life_and_refuses_the_short_one():
with on_day("2026-11-02T09:00:00+00:00"):
fefo_world()
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-7311"}))
pl = must(api("GET", f"/deliveries/{d['document_id']}/pick-list", OPERATOR))
picks = (pl["lines"] if "lines" in pl else pl["pick_list"])[0]["proposals"]
assert [(p["location_id"], p["lot_code"]) for p in picks] == [("WH1/A/02", "L-2612")] # L-2611 has 29 days, L-2601 is expired
body = refused(api("POST", f"/deliveries/{d['document_id']}/pick", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "lot_code": "L-2611", "quantity": "24", "unit_code": "C62"}]}), 409, "needs 60 days")
assert body["detail"]["gate"] == "shelf_life" and "L-2611 expires 2026-12-01" in body["detail"]["reason"]
refused(api("POST", f"/deliveries/{d['document_id']}/pick", OPERATOR, {"override_reason": "customer agreed", "picks": [
{"line_no": 1, "location_id": "WH1/A/01", "lot_code": "L-2611", "quantity": "24", "unit_code": "C62"}]}), 403, "inv:reservation:write")
ok = must(api("POST", f"/deliveries/{d['document_id']}/pick", MANAGER, {"override_reason": "customer agreed", "picks": [
{"line_no": 1, "location_id": "WH1/A/01", "lot_code": "L-2611", "quantity": "24", "unit_code": "C62"}]}))
assert ok["picks"][0]["lot_code"] == "L-2611"
expired = refused(api("POST", f"/deliveries/{d['document_id']}/pick", MANAGER, {"override_reason": "x", "picks": [
{"line_no": 1, "location_id": "WH1/A/03", "lot_code": "L-2601", "quantity": "24", "unit_code": "C62"}]}), 409, "expired on 2026-10-15")
assert expired["detail"]["gate"] == "shelf_life"
def test_inv39_an_expired_lot_is_not_available_and_fifo_proposes_the_oldest_received():
with on_day("2026-11-02T09:00:00+00:00"):
fefo_world(strategy="fifo")
av = must(api("GET", "/availability/COLA-075", OPERATOR))
assert av["warehouses"][0]["on_hand"] in ("54", "63") and av["warehouses"][0]["available"] == "54" # 18 + 36 usable; L-2601 (9) expired
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-7311"}))
pl = must(api("GET", f"/deliveries/{d['document_id']}/pick-list", OPERATOR))
picks = (pl["lines"] if "lines" in pl else pl["pick_list"])[0]["proposals"]
assert picks[0]["lot_code"] in ("L-2612", "L-2611") and picks[0]["lot_code"] != "L-2601"
def deliver(order, lot, qty_c62, location="WH1/A/01", bp=7311):
fakes.add_order(order, {1: {"product_id": "COLA-075", "unit_code": "C62", "quantity": qty_c62, "unit_numerator": 3, "unit_denominator": 4}})
fakes.ORDERS[order]["bp_id"] = bp
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": order}))
return must(api("POST", f"/deliveries/{d['document_id']}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": location, "lot_code": lot, "quantity": qty_c62, "unit_code": "C62"}]}))
def test_inv41_a_recall_names_every_customer_and_blocking_holds_the_lot_wherever_it_is():
world.catalogue()
world.warehouse(extra_bins=[("DAMAGED", {"stock_status": "blocked"})])
must(api("POST", "/settings", OWNER, {"key": "lots_enabled", "value": "true"}))
must(api("POST", "/stocked-products", MANAGER, {"product_id": "COLA-075", "tracking": "lot", "removal_strategy": "fefo"}))
world.receive("COLA-075", "8", "PK24", price="21.60", location="WH1/A/01", lot_code="L-2611", expires_on="2027-06-01") # 144 LTR
deliver("SO-1", "L-2611", "72") # 54 LTR to 7311
deliver("SO-2", "L-2611", "48", bp=8420) # 36 LTR to 8420
t = must(api("GET", "/lots/COLA-075/L-2611/trace", OPERATOR))
assert t["received_base"] == "144" and t["from"][0]["quantity_base"] == "144"
assert {c["bp_id"]: c["quantity_base"] for c in t["to"]} == {"7311": "54", "8420": "36"}
assert t["on_hand"] == [{"location_id": "WH1/A/01", "stock_type": "unrestricted", "quantity_base": "54"}] and t["unaccounted_base"] == "0"
held = must(api("POST", "/lots/COLA-075/L-2611/block", MANAGER, {"reason": "supplier recall"}))
assert held["blocked"] is True and held["changed"][0]["to_type"] == "blocked"
av = must(api("GET", "/availability/COLA-075", OPERATOR))
assert av["warehouses"][0]["on_hand"] == "54" and av["warehouses"][0]["available"] == "0" # held, still valued, never promised
fakes.add_order("SO-3", {1: {"product_id": "COLA-075", "unit_code": "C62", "quantity": "12", "unit_numerator": 3, "unit_denominator": 4}})
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-3"}))
refused(api("POST", f"/deliveries/{d['document_id']}/pick", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "lot_code": "L-2611", "quantity": "12", "unit_code": "C62"}]}), 409, "holds 0 LTR")
refused(api("POST", "/lots/COLA-075/L-2611/block", OPERATOR, {"reason": "x"}), 403, "inv:stock_type:write")
must(api("POST", "/lots/COLA-075/L-2611/release", MANAGER, {"reason": "cleared"}))
assert must(api("GET", "/availability/COLA-075", OPERATOR))["warehouses"][0]["available"] == "54"
def test_inv40_a_serial_has_a_history_and_is_on_hand_in_one_place_or_nowhere():
world.catalogue()
world.warehouse(bins=("A/03", "A/04"))
must(api("POST", "/settings", OWNER, {"key": "lots_enabled", "value": "true"}))
must(api("POST", "/stocked-products", MANAGER, {"product_id": "DRILL-500", "tracking": "serial"}))
world.receive("DRILL-500", "1", "C62", price="80.00", location="WH1/A/03", serials=["SN-1"])
world.receive("DRILL-500", "1", "C62", price="80.00", location="WH1/A/04", serials=["SN-2"])
h = must(api("GET", "/serials/DRILL-500/SN-1", OPERATOR))
assert h["on_hand_at"] == "WH1/A/03" and [e["event"] for e in h["history"]] == ["received"] and h["history"][0]["bp_id"] == "9100"
fakes.add_order("SO-9", {1: {"product_id": "DRILL-500", "unit_code": "C62", "quantity": "1"}})
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-9"}))
must(api("POST", f"/deliveries/{d['document_id']}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/03", "lot_code": "SN-1", "quantity": "1", "unit_code": "C62"}]}))
h = must(api("GET", "/serials/DRILL-500/SN-1", OPERATOR))
assert h["on_hand_at"] is None and [e["event"] for e in h["history"]] == ["received", "delivered"] and h["history"][1]["bp_id"] == "7311"
refused(api("GET", "/serials/DRILL-500/NOPE", OPERATOR), 404, "no serial")
refused(api("GET", "/serials/DRILL-500/SN-1", SYSTEM_NO_READ()), 404, "not found")
def SYSTEM_NO_READ():
from conftest import SALES
return SALES