dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_consignment.py
"""Supplier consignment: INV-46 (owner on the stock row, valued at zero, withdrawn at purchasing's price)."""
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, SYSTEM, api, must, refused
def stock():
return must(api("GET", "/stock/SCREW-M4", OPERATOR))["warehouses"][0]
def consigned():
world.catalogue()
world.warehouse(bins=("B/01",), zones=("B",))
fakes.add_po("PO-0150", {1: dict(product_id="SCREW-M4", unit_code="C62", ordered="5000", price="2.45", source_kind="consignment_receipt",
stock_owner_bp_id=9100)})
must(api("POST", "/receipts/scan", OPERATOR, {"purchase_order_id": "PO-0150", "delivery_note": "DN-C1", "location_id": "WH1/B/01",
"lines": [{"line_no": 1, "quantity": "5000", "unit_code": "C62"}]}))
def test_inv46_consigned_stock_is_valued_at_zero_usable_and_withdrawn_once_at_the_consignment_price_when_delivered():
consigned()
st = stock()
assert st["on_hand"] == "5000" and st["consignment"] == "5000"
assert must(api("GET", "/availability/SCREW-M4", OPERATOR))["warehouses"][0]["available"] == "5000" # we may use it
assert must(api("GET", "/valuation/products/SCREW-M4", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "0.00"
assert not [j for j in fakes.JOURNALS.values() if j["event"] == "receipt"] # not ours: no journal
fakes.add_order("SO-C", {1: {"product_id": "SCREW-M4", "unit_code": "C62", "quantity": "1200"}})
must(api("POST", "/reservations/sales-orders/SO-C", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [
{"line_no": 1, "product_id": "SCREW-M4", "unit_code": "C62", "quantity": "1200"}]}))
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-C"}))
must(api("POST", f"/deliveries/{d['document_id']}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/B/01", "quantity": "1200", "unit_code": "C62"}]}))
assert list(fakes.WITHDRAWALS)[-1].startswith("inv:") and fakes.WITHDRAWALS[[k for k in fakes.WITHDRAWALS if k != "_price"][0]]["amount"] == "29.40"
(w,) = [j for j in fakes.JOURNALS.values() if j["event"] == "consignment_withdrawal"]
assert {(l["purpose"], "debit" in l) for l in w["lines"]} == {("stock_valuation", True), ("consignment_liability", False)}
st = stock()
assert st["on_hand"] == "3800" and st["consignment"] == "3800" # 5000 - 1200 delivered, all from 9100's pile
cogs = [j for j in fakes.JOURNALS.values() if j["event"] == "delivery"][0]
assert any(l.get("debit") == "29.40" for l in cogs["lines"]) # delivered as our stock at the new average
def test_inv46_a_withdrawal_with_no_delivery_is_idempotent_and_a_missing_price_refuses_it():
consigned()
body = {"product_id": "SCREW-M4", "location_id": "WH1/B/01", "quantity": "300", "unit_code": "C62", "request_id": "w1"}
first = must(api("POST", "/consignment-stock/withdrawals", MANAGER, body))
assert first["withdrawn"][0]["amount"] == "7.35" and first["withdrawn"][0]["owner_bp_id"] == "9100"
assert must(api("POST", "/consignment-stock/withdrawals", MANAGER, body))["replay"] is True
assert stock()["consignment"] == "4700"
assert must(api("GET", "/valuation/products/SCREW-M4", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "7.35"
refused(api("POST", "/consignment-stock/withdrawals", MANAGER, {**body, "quantity": "9999", "request_id": "w2"}), 409, "owned by a supplier")
fakes.WITHDRAWALS["_price"] = None
body = refused(api("POST", "/consignment-stock/withdrawals", MANAGER, {**body, "quantity": "10", "request_id": "w3"}), 409, "no consignment price")
assert body["detail"]["gate"] == "consignment_price"
assert stock()["consignment"] == "4700" # nothing moved
refused(api("POST", "/consignment-stock/withdrawals", fakes_sales(), {"product_id": "SCREW-M4", "location_id": "WH1/B/01", "quantity": "1", "unit_code": "C62"}), 403, "missing permission")
def fakes_sales():
from conftest import SALES
return SALES
def test_inv47_components_sent_to_a_subcontractor_stay_ours_and_are_consumed_into_the_kit_at_components_plus_service():
world.catalogue()
fakes.add_product("KIT-A", base="C62", units={})
fakes.add_product("BOX-S", base="C62", units={})
world.warehouse(bins=("A/01", "A/09"))
world.receive("BOX-S", "100", "C62", price="1.00", location="WH1/A/01") # 100.00
world.receive("SCREW-M4", "200", "C62", price="0.10", location="WH1/A/01") # 20.00
body = {"move_ref": "pur:PO-0160:1:prov:1", "purchase_order_id": "PO-0160", "line_no": 1, "bp_id": "9200", "from_location_id": "WH1/A/01",
"provided_on": "2026-11-20", "components": [{"product_id": "BOX-S", "quantity": "100", "unit_code": "C62"},
{"product_id": "SCREW-M4", "quantity": "200", "unit_code": "C62"}]}
short = refused(api("POST", "/subcontract-provisions", SYSTEM, {**body, "move_ref": "pur:PO-0160:1:prov:0", "components": [
{"product_id": "BOX-S", "quantity": "999", "unit_code": "C62"}]}), 409, "holds 100")
assert short["detail"]["gate"] == "on_hand"
first = must(api("POST", "/subcontract-provisions", SYSTEM, body))
assert first["replay"] is False and [c["quantity_base"] for c in first["components"]] == ["100", "200"]
assert must(api("POST", "/subcontract-provisions", SYSTEM, body))["replay"] is True
assert must(api("GET", "/availability/BOX-S", OPERATOR))["warehouses"] == [] # at the subcontractor: ours, not available
assert must(api("GET", "/valuation/products/BOX-S", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "100.00"
fakes.add_po("PO-0160", {1: dict(product_id="KIT-A", unit_code="C62", ordered="100", price="0", source_kind="subcontract_receipt", service_fee="3.0000",
consume_provided=[{"product_id": "BOX-S", "quantity": "100", "unit_code": "C62"},
{"product_id": "SCREW-M4", "quantity": "200", "unit_code": "C62"}])}, bp_id=9200)
must(api("POST", "/receipts/scan", OPERATOR, {"purchase_order_id": "PO-0160", "delivery_note": "DN-S1", "location_id": "WH1/A/09",
"lines": [{"line_no": 1, "quantity": "100", "unit_code": "C62"}]}))
assert must(api("GET", "/valuation/products/KIT-A", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "420.00" # 120 of components + 300 of service
assert must(api("GET", "/valuation/products/BOX-S", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "0.00"
again = must(api("POST", "/receipts/scan", OPERATOR, {"purchase_order_id": "PO-0160", "delivery_note": "DN-S1", "location_id": "WH1/A/09",
"lines": [{"line_no": 1, "quantity": "100", "unit_code": "C62"}]}))
assert must(api("GET", "/valuation/products/KIT-A", ACCOUNTANT, {"company_code": "UK1"}))["value_amount"] == "420.00" and again is not None
def test_inv45_stock_at_a_customer_stays_ours_is_filled_issued_and_picked_up_and_never_available():
from decimal import Decimal
import fakes
import world
from conftest import ACCOUNTANT, MANAGER, OPERATOR, api, must, refused
world.catalogue()
world.warehouse()
world.receive("COLA-075", "10", "PK24", price="21.60", location="WH1/A/01") # 180 litres
bottles = {"product_id": "COLA-075", "unit_code": "C62", "unit_numerator": 3, "unit_denominator": 4}
fakes.add_order("CF-1", {1: {**bottles, "quantity": "240"}}, special_stock="to_consignment", purpose="consignment")
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "CF-1", "warehouse_id": "WH1"}))["document_id"]
done = must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": "240", "unit_code": "C62"}]}))
assert done["moves"][0]["to_location_id"] == "UK1:CONSIGNMENT:7311" and done["moves"][0]["quantity_base"] == "180"
assert must(api("GET", f"/documents/{doc}", ACCOUNTANT))["moves"][0]["value_amount"] == "0.00" and ("CF-1", done["delivery_ref"], 1) not in fakes.DELIVERY_REFS # no value change, not a delivery
assert not any(j["event"] == "delivery" for j in fakes.JOURNALS.values()) # no cost of sales
assert must(api("GET", "/consignment-stock/customers/7311", OPERATOR))["products"][0]["on_hand_base"] == "180"
assert all(w["available"] == "0" for w in must(api("GET", "/availability/COLA-075", OPERATOR))["warehouses"]) # customer stock is never available
# the customer uses 60 bottles: cost of sales at the average, and the order hears of it
fakes.add_order("CI-1", {1: {**bottles, "quantity": "60"}}, special_stock="from_consignment", purpose="consignment")
doc2 = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "CI-1", "warehouse_id": "WH1"}))["document_id"]
issued = must(api("POST", f"/deliveries/{doc2}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "UK1:CONSIGNMENT:7311", "quantity": "60", "unit_code": "C62"}]}))
assert issued["moves"][0]["from_location_id"] == "UK1:CONSIGNMENT:7311" and issued["moves"][0]["to_location_id"] == "UK1:CUSTOMER"
assert issued["moves"][0]["quantity_base"] == "45" and fakes.ORDERS["CI-1"]["lines"][1]["delivered"] == Decimal(60)
# pick-up of 24 bottles (18 litres) back to WH1/A/02
back = must(api("POST", "/consignment-stock/pickups", MANAGER, {"bp_id": 7311, "product_id": "COLA-075", "quantity": "24", "unit_code": "C62",
"to_location_id": "WH1/A/02", "request_id": "p1"}))
assert back["quantity_base"] == "18" and back["to_location_id"] == "WH1/A/02"
assert must(api("POST", "/consignment-stock/pickups", MANAGER, {"bp_id": 7311, "product_id": "COLA-075", "quantity": "24", "unit_code": "C62",
"to_location_id": "WH1/A/02", "request_id": "p1"}))["replay"] is True
refused(api("POST", "/consignment-stock/pickups", MANAGER, {"bp_id": 7311, "product_id": "COLA-075", "quantity": "999", "unit_code": "C62",
"to_location_id": "WH1/A/02", "request_id": "p2"}), 409, "holds less")
got = must(api("GET", "/consignment-stock/customers/7311", OPERATOR))
assert got["products"][0]["on_hand_base"] == "117" # 180 - 45 - 18
assert must(api("GET", "/stock/COLA-075", ACCOUNTANT))["on_hand"] == "18" # only the 18 picked back is in our warehouse