dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_audit1.py
"""Audit 1 findings, each proven by a test that failed first."""
from decimal import Decimal
import fakes
import world
from conftest import ACCOUNTANT, BOSS, MANAGER, OPERATOR, OWNER, SALES, SYSTEM, api, must, on_day, refused
BOTTLES = {"product_id": "COLA-075", "unit_code": "C62", "quantity": "48", "unit_numerator": 3, "unit_denominator": 4}
def test_inv50_goods_in_is_on_hand_but_not_available_nor_pickable_with_two_step_receipts():
world.catalogue()
world.warehouse(extra_bins=[("IN", {})])
must(api("POST", "/settings", BOSS, {"key": "receipt_steps", "value": "two"}))
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/IN") # 36 LTR in goods-in
av = must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]
assert (av["on_hand"], av["available"]) == ("36", "0")
atp = must(api("GET", "/atp/COLA-075", SALES, {"warehouse_id": "WH1", "quantity": "10", "unit_code": "LTR"}))
assert atp["available_now"] == "0"
fakes.add_order("SO-1", {1: BOTTLES})
got = must(api("POST", "/reservations/sales-orders/SO-1", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}))
assert got["lines"][0]["reserved_base"] == "0"
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-1", "warehouse_id": "WH1"}))
refused(api("POST", f"/deliveries/{d['document_id']}/pick", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/IN", "quantity": "12", "unit_code": "C62"}]}), 409, "put it away first")
world.receive("COLA-075", "1", "PK24", price="21.60", location="WH1/A/01") # shelf stock is available
assert must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]["available"] == "18"
def test_sc_inv_02_an_unreserved_delivery_cannot_take_stock_another_order_reserved():
world.catalogue()
world.warehouse()
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01") # 36 LTR
fakes.add_order("SO-1", {1: BOTTLES})
fakes.add_order("SO-2", {1: BOTTLES})
must(api("POST", "/reservations/sales-orders/SO-1", SYSTEM, {"revision": 1, "warehouse_id": "WH1", "lines": [{"line_no": 1, **BOTTLES}]}))
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-2", "warehouse_id": "WH1"}))
refused(api("POST", f"/deliveries/{d['document_id']}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "quantity": "48", "unit_code": "C62"}]}), 409, "reserved")
assert must(api("GET", "/stock/COLA-075", OPERATOR))["on_hand"] == "36"
def test_inv40_a_serial_received_in_two_places_at_once_collides_on_its_claim(monkeypatch):
from brick_inventory.application import ledger
from brick_inventory.infrastructure.tables import Repository
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"])
with monkeypatch.context() as m:
# a concurrent receiver whose snapshot predates the first receipt: it sees the serial nowhere and no claim yet
m.setattr(ledger, "serial_where", lambda unit, product_id, serial: "")
real = Repository.last_claim_no
m.setattr(Repository, "last_claim_no", lambda self, key: 0 if key[1] == "serial" else real(self, key))
world.purchase("PO-X", "DRILL-500", "5", "C62", "80.00")
status, body = api("POST", "/receipts/scan", OPERATOR, {"purchase_order_id": "PO-X", "delivery_note": "DN-X", "location_id": "WH1/A/04",
"lines": [{"line_no": 1, "quantity": "1", "unit_code": "C62", "serials": ["SN-1"]}]})
assert status == 409, (status, body)
assert must(api("GET", "/serials/DRILL-500/SN-1", OPERATOR))["on_hand_at"] == "WH1/A/03"
def test_inv41_stock_of_a_blocked_lot_received_later_lands_blocked_until_released():
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": "fefo"}))
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01", lot_code="L-1", expires_on="2027-06-01")
must(api("POST", "/lots/COLA-075/L-1/block", MANAGER, {"reason": "recall"}))
world.receive("COLA-075", "1", "PK24", price="21.60", location="WH1/A/02", lot_code="L-1", expires_on="2027-06-01") # a late delivery of the lot
av = must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]
assert (av["on_hand"], av["available"]) == ("54", "0")
fakes.add_order("SO-3", {1: BOTTLES})
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/02", "lot_code": "L-1", "quantity": "12", "unit_code": "C62"}]}), 409, "holds 0")
must(api("POST", "/lots/COLA-075/L-1/release", MANAGER, {"reason": "cleared"}))
assert must(api("GET", "/availability/COLA-075", SALES))["warehouses"][0]["available"] == "54"
def test_inv31_value_is_zero_when_a_receipt_brings_negative_stock_back_to_exactly_zero():
from brick_inventory.domain import valuation as val
prev = val.Position(Decimal("-2"), Decimal("-2.50"), Decimal("1.2500"), 3) # sold 2 we did not have, at the last average
step = val.average_step(prev, Decimal("2"), value_in=Decimal("3.00"))
assert step.after.quantity == 0 and step.after.value == 0 and step.value_change == Decimal("2.50")
def test_bounds_a_nonsense_hand_reservation_quantity_is_422_not_500():
world.catalogue()
world.warehouse()
for bad in ("sNaN", "NaN", "Infinity", "1e999999"):
status, body = api("POST", "/reservations/hand", MANAGER, {"sales_order_id": "SO-1", "line_no": 1, "quantity_base": bad,
"reason": "x", "product_id": "COLA-075"})
assert status == 422, (bad, status, body)
def test_bounds_other_string_quantities_are_422_too():
world.catalogue()
world.warehouse()
for path, body in (("/stock/change-type", {"location_id": "WH1/A/01", "product_id": "COLA-075", "unit_code": "LTR", "from_type": "unrestricted",
"to_type": "blocked", "reason": "x"}),
("/consignment-stock/withdrawals", {"product_id": "COLA-075", "location_id": "WH1/A/01", "unit_code": "LTR"}),
("/consignment-stock/pickups", {"bp_id": 1, "product_id": "COLA-075", "to_location_id": "WH1/A/01", "unit_code": "LTR"})):
for bad in ("sNaN", "NaN", "1e999999"):
status, out = api("POST", path, MANAGER, {**body, "quantity": bad})
assert status == 422, (path, bad, status, out)
for bad in ("sNaN", "1e999999"):
status, out = api("POST", "/locations", MANAGER, {"location_id": "WH1/A/09", "kind": "internal", "max_quantity_base": bad})
assert status == 422, (bad, status, out)
def test_inv10_a_count_applied_after_a_delivery_took_stock_from_the_bin_takes_the_difference_once():
world.catalogue()
world.warehouse(zones=("A",))
with on_day("2026-11-20T07:00:00+00:00"):
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01") # 36 LTR
with on_day("2026-11-20T08:00:00+00:00"):
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A/01"}))["count_id"]
with on_day("2026-11-20T10:00:00+00:00"): # counted 30 against 36: -6
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [
{"location_id": "WH1/A/01", "product_id": "COLA-075", "quantity": "30", "unit_code": "LTR"}]}))
with on_day("2026-11-20T11:00:00+00:00"): # a delivery takes 9 LTR afterwards
fakes.add_order("SO-7", {1: BOTTLES | {"quantity": "12"}})
d = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": "SO-7", "warehouse_id": "WH1"}))
must(api("POST", f"/deliveries/{d['document_id']}/validate", OPERATOR, {"picks": [
{"line_no": 1, "location_id": "WH1/A/01", "quantity": "12", "unit_code": "C62"}]}))
with on_day("2026-11-20T12:00:00+00:00"):
sub = must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
assert sub["lines"][0]["difference_base"] == "-6", sub
must(api("POST", f"/counts/{cid}/apply", MANAGER, {}))
assert must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["on_hand"] == "21" # 36 - 9 delivered - 6 lost, once
def test_inv48_a_goods_issue_reversal_on_standard_cost_is_valued_at_the_posted_value():
world.catalogue()
world.warehouse(bins=("A/05",))
must(api("POST", "/settings", OWNER, {"key": "valuation_method:UK1", "value": "standard", "valid_from": "2026-12-01"}))
must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "GLOVES", "standard_price": "0.0800", "valid_from": "2026-12-01"}))
with on_day("2026-12-02T09:00:00+00:00"):
world.receive("GLOVES", "20", "BX100", price="8.00", location="WH1/A/05")
plan = must(api("POST", "/goods-issues", OPERATOR, {"purpose": "cost_centre", "cost_centre": "MAINT", "lines": [
{"product_id": "GLOVES", "quantity": "3", "from_location_id": "WH1/A/05"}]}))
must(api("POST", f"/goods-issues/{plan['document_id']}/post", OPERATOR, {}))
must(api("POST", "/standard-prices", ACCOUNTANT, {"company_code": "UK1", "product_id": "GLOVES", "standard_price": "0.1000", "valid_from": "2026-12-20"}))
with on_day("2026-12-20T09:00:00+00:00"):
must(api("POST", "/standard-prices/apply", SYSTEM, {}))
rev = must(api("POST", f"/goods-issues/{plan['document_id']}/reverse", MANAGER, {"reason": "wrong cost centre"}))
(mv,) = must(api("GET", f"/documents/{plan['document_id']}:rev", ACCOUNTANT))["moves"]
assert mv["value_amount"] == "24.00", (mv, rev) # as posted, not 300 x 0.1000