dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_transfers.py
"""Transfers: INV-09 between bins and sites, INV-15 reverse a mistake, INV-36 release from quality, INV-53 expected arrival."""
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, SALES, api, must, refused
def two_sites():
world.catalogue()
world.warehouse(bins=("A/01", "B/07"), zones=("A", "B"), extra_bins=[("QC", {"stock_status": "quality"})])
world.warehouse("WH2", bins=("A/01",))
world.receive("COLA-075", "2", "PK24", location="WH1/A/01") # 36 LTR
def plan(src, dst, qty="1", unit="PK24", **kw):
return must(api("POST", "/transfers", MANAGER, {"from_location_id": src, "to_location_id": dst,
"lines": [{"product_id": "COLA-075", "quantity": qty, "unit_code": unit, **kw}]}))
def test_inv09_a_move_between_bins_changes_no_value():
two_sites()
t = plan("WH1/A/01", "WH1/B/07", "1")
done = must(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}))
assert done["number"] == "WH1/INT/00001" and done["lifecycle"] == "done"
stock = must(api("GET", "/stock/COLA-075", OPERATOR))
assert {l["location_id"]: l["quantity_base"] for l in stock["warehouses"][0]["locations"]} == {"WH1/A/01": "18", "WH1/B/07": "18"}
val = must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1"}))
assert val["value_amount"] == "43.20" and len(val["valuation_rows"]) == 1 # a transfer appends no valuation row
assert must(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}))["replay"] is True
def test_inv09_a_move_to_another_site_goes_through_transit_and_back_in_nobodys_available():
two_sites()
t = plan("WH1/A/01", "WH2/A/01", "1")
refused(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}), 409, "different warehouses")
shipped = must(api("POST", f"/transfers/{t['document_id']}/ship", OPERATOR, {}))
assert shipped["lifecycle"] == "in_transit" and shipped["expected_arrival_on"] is None
stock = must(api("GET", "/stock/COLA-075", OPERATOR))
assert stock["in_transit"] == "18" and stock["on_hand"] == "18"
av = {w["warehouse_id"]: w for w in must(api("GET", "/availability/COLA-075", SALES))["warehouses"]}
assert av["WH1"]["available"] == "18" and "WH2" not in av or av["WH2"]["available"] == "0"
assert av["WH2" if "WH2" in av else "WH1"]["in_transit_in"] in ("18", "0")
got = must(api("POST", f"/transfers/{t['document_id']}/receive", OPERATOR, {}))
assert got["lifecycle"] == "done" and got["shortfall"] == [] and got["readiness"][0]["met"] is True
stock = must(api("GET", "/stock/COLA-075", OPERATOR))
assert {w["warehouse_id"]: w["on_hand"] for w in stock["warehouses"]} == {"WH1": "18", "WH2": "18"} and stock["in_transit"] == "0"
def test_inv09_another_company_is_a_sale_and_a_purchase_not_a_move():
two_sites()
world.warehouse("WH9", company="DE1", bins=("A/01",))
refused(api("POST", "/transfers", MANAGER, {"from_location_id": "WH1/A/01", "to_location_id": "WH9/A/01",
"lines": [{"product_id": "COLA-075", "quantity": "1", "unit_code": "PK24"}]}),
409, "inter-company move is a sale and a purchase")
def test_inv53_shipping_sets_the_expected_arrival_and_a_short_receipt_stays_in_transit():
two_sites()
fakes.SITES.append({"product_id": "COLA-075", "site_id": "WH2", "supplying_site_id": "WH1", "planned_delivery_days": 2})
t = plan("WH1/A/01", "WH2/A/01", "2")
shipped = must(api("POST", f"/transfers/{t['document_id']}/ship", OPERATOR, {}))
assert shipped["expected_arrival_on"] == "2026-11-22"
refused(api("POST", f"/transfers/{t['document_id']}/receive", OPERATOR, {"lines": [{"line_no": 1, "quantity": "3", "unit_code": "PK24"}]}), 409, "only 36 was shipped")
got = must(api("POST", f"/transfers/{t['document_id']}/receive", OPERATOR, {"lines": [{"line_no": 1, "quantity": "1", "unit_code": "PK24"}]}))
assert got["shortfall"] == [{"line_no": 1, "product_id": "COLA-075", "short_base": "18"}] and got["readiness"][0]["met"] is False
assert must(api("GET", "/stock/COLA-075", OPERATOR))["in_transit"] == "18"
assert must(api("GET", "/transfers", OPERATOR, {"overdue": True}))["transfers"] == []
from conftest import on_day
with on_day("2026-11-25T09:00:00+00:00"):
assert must(api("GET", "/transfers", OPERATOR, {"lifecycle": "in_transit", "overdue": True}))["transfers"] == [] # done: not overdue
def test_inv36_goods_in_quality_are_on_hand_and_valued_but_never_available_until_released():
two_sites()
world.receive("COLA-075", "8", "PK24", location="WH1/QC", note="DN-QC")
av = must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]
assert (av["on_hand"], av["available"]) == ("180", "36") # 36 unrestricted + 144 in QC
rows = must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]
assert rows["by_type"] == {"unrestricted": "36", "quality": "144", "blocked": "0"}
rel = plan("WH1/QC", "WH1/A/01", "6")
must(api("POST", f"/transfers/{rel['document_id']}/move", OPERATOR, {}))
av = must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]
assert av["available"] == "144" # 36 + 108 released
assert must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "216.00"
fakes.add_order("SO-Q", {1: {"product_id": "COLA-075", "unit_code": "PK24", "quantity": "1", "unit_numerator": 18, "unit_denominator": 1}})
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-Q", "warehouse_id": "WH1"}))["document_id"]
refused(api("POST", f"/deliveries/{doc}/pick", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/QC", "quantity": "1", "unit_code": "PK24"}]}),
409, "WH1/QC is in quality inspection: release it first")
def test_inv55_hold_stock_where_it_stands_and_release_it():
two_sites()
must(api("POST", "/stock/change-type", MANAGER, {"location_id": "WH1/A/01", "product_id": "COLA-075", "quantity": "1", "unit_code": "PK24",
"from_type": "unrestricted", "to_type": "quality", "reason": "doubtful pallet"}))
av = must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]
assert (av["on_hand"], av["available"]) == ("36", "18")
must(api("POST", "/stock/change-type", MANAGER, {"location_id": "WH1/A/01", "product_id": "COLA-075", "quantity": "1", "unit_code": "PK24",
"from_type": "quality", "to_type": "blocked", "reason": "failed test"}))
row = {r["stock_type"]: r["quantity_base"] for r in must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["locations"]}
assert row == {"unrestricted": "18", "blocked": "18"}
refused(api("POST", "/stock/change-type", MANAGER, {"location_id": "WH1/QC", "product_id": "COLA-075", "quantity": "1", "unit_code": "PK24",
"from_type": "quality", "to_type": "unrestricted", "reason": "x"}), 409, "takes quality stock only")
refused(api("POST", "/stock/change-type", OPERATOR, {"location_id": "WH1/A/01", "product_id": "COLA-075", "quantity": "1", "unit_code": "PK24",
"from_type": "unrestricted", "to_type": "quality", "reason": "x"}), 403, "inv:stock_type:write")
assert must(api("GET", "/valuation/products/COLA-075", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "43.20" # no value change
def test_inv15_a_mistake_is_reversed_by_a_move_that_names_it_and_only_once():
two_sites()
t = plan("WH1/A/01", "WH1/B/07", "1")
must(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}))
move_id = must(api("GET", f"/documents/{t['document_id']}", OPERATOR))["moves"][0]["move_id"]
got = must(api("POST", f"/moves/{move_id}/reverse", MANAGER, {"reason": "wrong bin"}))
assert got["reverses_move_id"] == move_id
moves = must(api("GET", "/moves", OPERATOR, {"product_id": "COLA-075"}))["moves"]
assert [m["reverses_move_id"] for m in moves if m["reverses_move_id"]] == [move_id]
refused(api("POST", f"/moves/{move_id}/reverse", MANAGER, {"reason": "again"}), 409, "already reversed")
first = must(api("GET", "/documents/rcpt:pur:R-1", OPERATOR))["moves"][0]["move_id"]
refused(api("POST", f"/moves/{first}/reverse", MANAGER, {"reason": "x"}), 409, "a delivery is undone by a return")
stock = must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["locations"]
assert {l["location_id"]: l["quantity_base"] for l in stock} == {"WH1/A/01": "36"}