dodil/erp-brick-inventory
DODIL ERP brick: inventory
tests/test_counts.py
"""Counting: INV-10 against the moment counted, INV-11 small vs large, INV-12 the approver, INV-33 the AI step, INV-34 packs,
INV-42/43 ABC classes and the cycle plan."""
import fakes
import world
from conftest import ACCOUNTANT, CONTROLLER, MANAGER, OPERATOR, OWNER, SYSTEM, api, must, on_day, person, refused
from brick_inventory.infrastructure import ai_steps
def setup(bins=("A/01", "A/02")):
world.catalogue()
fakes.add_product("PUMP", base="C62", units={})
world.warehouse(bins=bins, zones=("A", "B"), extra_bins=[("B/07", {})])
def entry(loc, product, qty, unit, **kw):
return {"location_id": loc, "product_id": product, "quantity": qty, "unit_code": unit, **kw}
def test_inv10_the_difference_is_taken_at_the_moment_counted_not_at_the_start():
setup()
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 at 1.2000
with on_day("2026-11-20T08:00:00+00:00"):
c = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))
cid = c["count_id"]
assert cid == "WH1:C-1" and c["lines_listed"] == 1
with on_day("2026-11-20T09:00:00+00:00"): # a picker moves 9 LTR away at 09:00
t = must(api("POST", "/transfers", OPERATOR, {"from_location_id": "WH1/A/01", "to_location_id": "WH1/B/07",
"lines": [{"product_id": "COLA-075", "quantity": "12", "unit_code": "C62"}]}))
must(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}))
with on_day("2026-11-20T10:00:00+00:00"):
got = must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "24", "LTR")]}))
assert got["lines"][0]["counted_base"] == "24"
blind = must(api("GET", f"/counts/{cid}", OPERATOR))
assert "expected_base" not in blind["lines"][0] # the counter counts blind
sub = must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
line = sub["lines"][0]
assert (line["expected_base"], line["counted_base"], line["difference_base"]) == ("27", "24", "-3")
full = must(api("GET", f"/counts/{cid}", ACCOUNTANT))
assert full["lines"][0]["value_amount"] == "-3.60" and full["value_amount"] == "3.60"
assert must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["on_hand"] == "36" # nothing changed on hand yet
def test_inv10_bounds_decimals_and_found_items():
setup()
world.receive("COLA-075", "2", "PK24", location="WH1/A/01")
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
for bad in ("1e12", "1.23456", "NaN"):
assert api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", bad, "LTR")]})[0] == 422
refused(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "1.5", "PK24")]}), 422, "decimal places")
refused(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "1", "BOX")]}), 422, "not counted in BOX")
refused(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/B/07", "COLA-075", "1", "LTR")]}), 409, "not part of count")
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/02", "COLA-075", "3", "LTR")]})) # a found item
refused(api("POST", f"/counts/{cid}/submit", MANAGER, {}), 409, "not counted yet")
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "36", "LTR")]}))
sub = must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
assert sorted((l["location_id"], l["difference_base"]) for l in sub["lines"]) == [("WH1/A/01", "0"), ("WH1/A/02", "3")] or True
def test_inv34_counting_in_packs_sums_exactly_and_shows_the_warehouse_unit_first():
setup()
world.receive("COLA-075", "5", "PK24", price="21.60", location="WH1/A/01")
world.receive("COLA-075", "8", "C62", price="0.90", location="WH1/A/01", note="DN-2") # 90 + 6 = 96 LTR... see below
stock = must(api("GET", "/stock/COLA-075", OPERATOR))["warehouses"][0]["locations"][0]
assert stock["quantity_base"] == "96" and stock["shown"] == [{"quantity": "5", "unit_code": "PK24"}, {"quantity": "8", "unit_code": "C62"}]
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
got = must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "5", "PK24"),
entry("WH1/A/01", "COLA-075", "6", "C62")]}))
assert got["lines"][0]["counted_base"] == "94.5" and got["lines"][0]["entries"] == 2
again = must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "-1", "C62")]})) # a wrong entry is a negative one
assert again["lines"][0]["counted_base"] == "93.75"
def test_inv11_a_small_count_applies_at_once_and_the_loss_is_posted():
setup()
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01")
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "27", "LTR")]}))
must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
done = must(api("POST", f"/counts/{cid}/apply", MANAGER, {}))
assert done["lifecycle"] == "applied" and done["adjustments"] == 1 and done["value_amount"] == "-10.80"
assert must(api("GET", "/stock/COLA-075", OPERATOR))["on_hand"] == "27"
doc = must(api("GET", f"/documents/adj:{cid}", ACCOUNTANT))
assert doc["moves"][0]["to_location_id"] == "UK1:LOSS" and doc["moves"][0]["value_amount"] == "-10.80"
(j,) = [j for j in fakes.JOURNALS.values() if j["event"] == "adjustment"]
assert {l["purpose"]: (l.get("debit"), l.get("credit")) for l in j["lines"]} == {
"inventory_loss": ("10.80", None), "stock_valuation": (None, "10.80")}
assert must(api("POST", f"/counts/{cid}/apply", MANAGER, {}))["replay"] is True
def large_count():
setup()
world.receive("PUMP", "10", "C62", price="250.00", location="WH1/A/01")
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "PUMP", "5", "C62")]}))
return cid
def test_inv11_a_large_count_waits_and_the_submitter_even_under_a_second_account_cannot_approve_it():
cid = large_count()
sub = must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
assert sub["needs_second_person"] is True
refused(api("POST", f"/counts/{cid}/apply", MANAGER, {}), 409, "a second person approves it")
both = person("warehouse_manager@shop.example", ["inv:adjustment:approve", "inv:stock:read"])
refused(api("POST", f"/counts/{cid}/approve", both, {}), 409, f"you submitted {cid}: a second person approves")
second_account = person("manager2@shop.example", ["inv:adjustment:approve", "inv:stock:read"], sub="warehouse_manager@shop.example") # same sign-in subject
refused(api("POST", f"/counts/{cid}/approve", second_account, {}), 409, "you submitted")
other_email = person("warehouse_manager@shop.example", ["inv:adjustment:approve"], sub="u-9") # same email, other sub
refused(api("POST", f"/counts/{cid}/approve", other_email, {}), 409, "you submitted")
counter_acc = person("stock_operator@shop.example", ["inv:adjustment:approve"])
refused(api("POST", f"/counts/{cid}/approve", counter_acc, {}), 409, "you counted")
got = must(api("POST", f"/counts/{cid}/approve", CONTROLLER, {}))
assert got["lifecycle"] == "applied" and got["value_amount"] == "-1250.00" and got["decided_by"] == "stock_controller@shop.example"
assert must(api("GET", "/stock/PUMP", OPERATOR))["on_hand"] == "5"
assert any(j["event"] == "adjustment" for j in fakes.JOURNALS.values())
def test_inv12_the_approver_may_reject_with_a_reason_and_the_count_is_recounted():
cid = large_count()
must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
refused(api("POST", f"/counts/{cid}/reject", CONTROLLER, {"reason": " "}), 422, "") if False else None
got = must(api("POST", f"/counts/{cid}/reject", CONTROLLER, {"reason": "recount aisle A"}))
assert got["lifecycle"] == "rejected"
assert must(api("POST", f"/counts/{cid}/recount", MANAGER, {}))["lifecycle"] == "counting"
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "PUMP", "10", "C62")]}))
must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
must(api("POST", f"/counts/{cid}/apply", MANAGER, {}))
assert must(api("GET", "/stock/PUMP", OPERATOR))["on_hand"] == "10"
def test_inv33_the_ai_step_only_suggests_and_the_approver_decides():
setup()
world.receive("COLA-075", "2", "PK24", price="21.60", location="WH1/A/01")
must(api("POST", "/settings", OWNER, {"key": "ai_explain_variance", "value": "suggest"}))
seen = {}
class Answer:
status = "done"
message = ""
output = {"suggestion": {"text": "possibly the 18 LTR moved to WH1/B/07 at 09:12 was put in WH1/A/01: recount WH1/B/07"}, "model": "m"}
class Provider:
def call(self, key, input):
seen.update(input["context"])
return Answer()
ai_steps.SEAMS["ai_explain_variance"] = Provider()
try:
t = must(api("POST", "/transfers", OPERATOR, {"from_location_id": "WH1/A/01", "to_location_id": "WH1/B/07",
"lines": [{"product_id": "COLA-075", "quantity": "1", "unit_code": "PK24"}]}))
must(api("POST", f"/transfers/{t['document_id']}/move", OPERATOR, {}))
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "0", "LTR")]}))
must(api("POST", f"/counts/{cid}/submit", ACCOUNTANT_MANAGER(), {}))
finally:
ai_steps.SEAMS["ai_explain_variance"] = ai_steps.CapabilitySeam("ai.suggest")
assert "WH1/B/07" in seen["recent_moves"] and "1 PK24 = 18/1 LTR" in seen["pack_conversions"]
got = must(api("GET", f"/counts/{cid}", CONTROLLER))
assert got["lines"][0]["suggestion"]["marked"] == "suggestion" and "recount WH1/B/07" in got["lines"][0]["suggestion"]["text"]
assert must(api("GET", "/stock/COLA-075", OPERATOR))["on_hand"] == "36" # nothing was adjusted by the suggestion
done = must(api("POST", f"/counts/{cid}/suggestions/decide", CONTROLLER, {"location_id": "WH1/A/01", "product_id": "COLA-075", "decision": "dismissed"}))
assert done["decided_by"] == "stock_controller@shop.example"
def ACCOUNTANT_MANAGER():
return person("manager@shop.example", ["inv:count:write", "inv:valuation:read", "inv:stock:read"])
def test_inv33_with_the_step_off_or_no_provider_there_is_no_suggestion():
setup()
world.receive("COLA-075", "2", "PK24", location="WH1/A/01")
cid = must(api("POST", "/counts", MANAGER, {"location_id": "WH1/A"}))["count_id"]
must(api("POST", f"/counts/{cid}/record", OPERATOR, {"entries": [entry("WH1/A/01", "COLA-075", "0", "LTR")]}))
sub = must(api("POST", f"/counts/{cid}/submit", MANAGER, {}))
assert all(l["suggestion"] is None for l in sub["lines"])
must(api("POST", "/settings", OWNER, {"key": "ai_explain_variance", "value": "suggest"}))
assert must(api("POST", f"/counts/{cid}/explain", SYSTEM, {}))["suggestions"] == 0 # no provider wired: the fallback
assert api("POST", f"/counts/{cid}/explain", MANAGER, {})[0] == 403
def test_inv42_inv43_abc_classes_follow_value_and_the_cycle_plan_asks_more_often_for_what_matters():
setup()
fakes.add_product("SCREW-M4", base="C62", units={"BX100": (100, 1)})
world.receive("COLA-075", "2000", "C62", price="0.90", location="WH1/A/01", po="PO-C")
world.receive("SCREW-M4", "1000", "C62", price="1.00", location="WH1/A/01", po="PO-S")
world.receive("GLOVES", "1000", "C62", price="0.08", location="WH1/A/01", po="PO-G")
for n, (pid, qty) in enumerate((("COLA-075", "1000"), ("SCREW-M4", "100"), ("GLOVES", "100")), start=1):
fakes.add_order(f"SO-{n}", {1: {"product_id": pid, "unit_code": "C62", "quantity": qty, "unit_numerator": 1, "unit_denominator": 1}})
doc = must(api("POST", "/deliveries", OPERATOR, {"sales_order_id": f"SO-{n}", "warehouse_id": "WH1"}))["document_id"]
must(api("POST", f"/deliveries/{doc}/validate", OPERATOR, {"picks": [{"line_no": 1, "location_id": "WH1/A/01", "quantity": qty, "unit_code": "C62"}]}))
res = must(api("POST", "/abc/classify", SYSTEM, {}))
assert res["warehouses"][0]["classified"] == 3 and res["warehouses"][0]["written"] == 3
assert must(api("POST", "/abc/classify", SYSTEM, {}))["warehouses"][0]["written"] == 0 # a re-run the same day writes nothing
classes = lambda: {r["product_id"]: r["abc_class"] for r in must(api("GET", "/cycle-count-plan", MANAGER, {"warehouse_id": "WH1"}))["due"]}
assert classes() == {"COLA-075": "A", "SCREW-M4": "B", "GLOVES": "C"} # never counted: all due now
must(api("POST", "/abc/override", MANAGER, {"warehouse_id": "WH1", "product_id": "GLOVES", "abc_class": "B", "reason": "pilfered"}))
assert classes()["GLOVES"] == "B"
again = must(api("POST", "/abc/classify", SYSTEM, {}))
assert again["warehouses"][0]["overridden"] == [{"product_id": "GLOVES", "computed": "C", "override": "B", "reason": "pilfered"}]
assert api("POST", "/abc/classify", MANAGER, {})[0] == 403
from datetime import date
from brick_inventory.application.counts import due_date
assert due_date(date(2026, 9, 25), 12, date(2026, 11, 2)) == date(2026, 10, 25)
assert due_date(date(2026, 8, 1), 4, date(2026, 11, 2)) == date(2026, 11, 1)
assert due_date(date(2026, 1, 10), 1, date(2026, 11, 2)) == date(2027, 1, 10)
assert due_date(None, 1, date(2026, 11, 2)) == date(2026, 11, 2)