dodil/erp-brick-product
tests/test_v4_slice_b.py
"""Product v4, slice B: templates and the composed unit (PRD-34), product types (PRD-35), characteristics and search (PRD-36),
site data (PRD-37), rounding profiles (PRD-38), the seam routes other bricks read (PRD-39)."""
from decimal import Decimal as D
from conftest import CATALOGUE, DEFAULT, SALES, SELLING, api, mk_list, must, on_day, price, row
PAST = "2026-11-19T09:00:00+00:00"
VOLUMES = [{"name": "volume", "values": [{"value": "0.33 L", "code": "033"}, {"value": "0.75 L", "code": "075"},
{"value": "1.5 L", "code": "150"}]}]
BASE = {"product_type_code": "LEGACY", "name": "Cola", "base_unit_code": "LTR", "tax_category": "standard", "billing_kind": "one_off", "invoicing_policy": "ordered"}
def units_of(pid, on=None):
return {u["unit_code"]: u for u in must(api("GET", f"/products/{pid}", CATALOGUE, {"on": on} if on else None))["units"]}
def cola_family():
"""COLA, base LTR: PK24 = 24 C62 on the TEMPLATE (the bottle C62 is per SKU: 0.33, 0.75, 1.5 LTR), gross 19.2 KGM, bought in 80s."""
must(api("POST", "/templates/COLA/generate-variants", CATALOGUE, {**BASE, "attributes": VOLUMES, "units_of": [
{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24", "packaging_level": "case", "einvoice_unit_code": "XPK"}]}))
for pid, (n, d) in {"COLA-033": (33, 100), "COLA-075": (3, 4), "COLA-150": (3, 2)}.items():
must(api("POST", f"/products/{pid}/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [
{"unit_code": "C62", "numerator": n, "denominator": d, "packaging_level": "each"}]}))
with on_day(PAST):
must(api("POST", "/products/COLA-075/set-measures", CATALOGUE, {"unit_code": "PK24", "gross_weight": "19.2", "weight_unit_code": "KGM", "scope": "template"}))
must(api("POST", "/products/COLA-075/set-rounding", CATALOGUE, {"role": "purchase", "unit_code": "PK24", "rounding_multiple": "80", "scope": "template"}))
def test_prd_34_a_template_pack_is_composed_with_each_skus_own_bottle_and_frozen_by_use():
"""PRD-34 — template COLA, PK24 = 24 C62 relative to the bottle, which differs per SKU; COLA-150 has its own measures and PK6; an order
for COLA-075 freezes the template's PK24; a weight correction is a new dated template row; the template takes no GTIN."""
cola_family()
must(api("POST", "/products/COLA-150/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "6", "einvoice_unit_code": "XPK"}]}))
must(api("POST", "/products/COLA-150/set-measures", CATALOGUE, {"unit_code": "PK24", "gross_weight": "37.4", "weight_unit_code": "KGM"}))
must(api("POST", "/products/COLA-150/set-unit-role", CATALOGUE, {"role": "sales", "unit_code": "PK6", "valid_from": "2026-11-21"}))
for pid, base in (("COLA-075", D(18)), ("COLA-150", D(36)), ("COLA-033", D("7.92"))): # 24 x 0.75, 24 x 1.5, 24 x 0.33: exact
u = units_of(pid)["PK24"]
assert D(u["numerator"]) / D(u["denominator"]) == base, (pid, u)
assert must(api("GET", "/products/COLA-075/convert", CATALOGUE, {"quantity": "2", "from_unit": "PK24", "to_unit": "C62"}))["converted"] == "48"
assert units_of("COLA-075")["PK24"]["measures"]["gross_weight"] == "19.2" and units_of("COLA-150")["PK24"]["measures"]["gross_weight"] == "37.4"
assert units_of("COLA-033")["PK24"]["measures"]["gross_weight"] == "19.2" # inherited whole
rule = must(api("GET", "/products/COLA-075/rounding", CATALOGUE, {"role": "purchase", "quantity": "100", "unit_code": "PK24", "on": "2026-11-21"}))
assert rule["rule"]["rounding_multiple"] == "80" and rule["met"] is False
assert must(api("GET", "/products/COLA-150", CATALOGUE, {"on": "2026-11-21"}))["roles"]["sales"] == "PK6"
assert must(api("GET", "/products/COLA-075", CATALOGUE, {"on": "2026-11-21"}))["roles"]["sales"] == "C62"
# the order for COLA-075 uses PK24: the template's row is frozen, naming what used it
must(api("POST", "/products/COLA-075/activate-product", CATALOGUE, {}))
must(api("POST", "/products/COLA-075/units/PK24/used", SELLING, {"document_ref": "sales order SO-2026-0012"}))
status, body = api("POST", "/templates/COLA/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "12"}]})
assert status == 409 and "PK24 on template COLA is used by sales order SO-2026-0012 for COLA-075" in body["detail"], body
assert api("POST", "/templates/COLA/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24", "packaging_level": "case"}]})[0] == 200
assert api("POST", "/products/COLA-150/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "8", "einvoice_unit_code": "XPK"}]})[0] == 200 # its own, unused
# a weight correction is a new dated template row; the order keeps the copy it took
must(api("POST", "/products/COLA-075/units/PK24/used", SELLING, {"document_ref": "sales order SO-2026-0012", "measures_from": "2026-11-19"}))
must(api("POST", "/products/COLA-075/set-measures", CATALOGUE, {"unit_code": "PK24", "gross_weight": "19.4", "scope": "template"}))
assert units_of("COLA-075")["PK24"]["measures"]["gross_weight"] == "19.4" and units_of("COLA-033")["PK24"]["measures"]["gross_weight"] == "19.4"
assert units_of("COLA-075", "2026-11-19")["PK24"]["measures"]["gross_weight"] == "19.2"
assert units_of("COLA-150")["PK24"]["measures"]["gross_weight"] == "37.4"
status, body = api("POST", "/products/COLA-075/add-gtin", CATALOGUE, {"unit_code": "PK24", "gtin": "05012345000015", "scope": "template"})
assert status == 409 and "COLA has 3 products: a GTIN names one, set it on the SKU" in body["detail"]
def sells(pid, ptype, **kw):
return must(api("POST", "/products", CATALOGUE, {"name": f"{ptype} thing", "product_type_code": ptype, "base_unit_code": "C62",
"tax_category": "standard", "billing_kind": "one_off", "invoicing_policy": "ordered",
**({"product_id": pid} if pid else {}), **kw}))
def test_prd_35_a_type_numbers_the_code_gates_activation_and_is_fixed_by_use():
"""PRD-35 — TRADING numbers TRD-00001 from its series (no gap from a refused add), needs a purchase unit to activate; RAW is not
sold; a type changes only while the product is an unused draft; inventory reads stocked_default."""
first = sells(None, "TRADING")
assert first["product_id"] == "TRD-00001" and first["kind"] == "goods"
assert api("POST", "/products", CATALOGUE, {"name": "x", "product_type_code": "TRADING", "base_unit_code": "ZZZ"})[0] == 422 # refused: no number spent
assert sells(None, "TRADING")["product_id"] == "TRD-00002"
got = must(api("GET", "/products/TRD-00001", SALES if False else CATALOGUE))
assert (got["product_type_code"], got["stocked_default"], got["valuation_class"]) == ("TRADING", True, "trading_goods")
must(api("POST", "/products/TRD-00001/set-units", CATALOGUE, {"sales_unit_code": "C62"}))
status, body = api("POST", "/products/TRD-00001/activate-product", CATALOGUE, {})
assert status == 409 and "TRADING needs a purchase unit" in body["detail"], body
must(api("POST", "/products/TRD-00001/set-units", CATALOGUE, {"purchase_unit_code": "C62"}))
assert must(api("POST", "/products/TRD-00001/activate-product", CATALOGUE, {}))["status"] == "active"
assert api("POST", "/products", SALES, {"name": "x", "product_type_code": "TRADING", "base_unit_code": "C62"})[0] == 403
assert api("POST", "/products", CATALOGUE, {"name": "x", "product_type_code": "NOPE", "base_unit_code": "C62"})[0] == 422
# RAW: bought, not sold, stocked; the line data says so
raw = sells("RAW-1", "RAW")
assert raw["kind"] == "goods" and must(api("GET", "/products/RAW-1", CATALOGUE))["stocked_default"] is True
must(api("POST", "/products/RAW-1/activate-product", CATALOGUE, {}))
mk_list("LIST-GBP", rows=[row("RAW-1", "2026-11-01", "1.00")], default_from=DEFAULT)
line = must(price("RAW-1"))["line"]
assert line["product_type_code"] == "RAW" and line["sellable"] is False and "RAW products are not sold" in line["sellable_reason"]
svc = sells("SVC-1", "SERVICE")
assert svc["kind"] == "service" and must(api("GET", "/products/SVC-1", CATALOGUE))["stocked_default"] is False
# a draft's type changes with a reason and the change is recorded; a used product's type is fixed
sells("DRAFT-1", "RAW")
assert api("POST", "/products/DRAFT-1/change-product-type", CATALOGUE, {"product_type_code": "TRADING"})[0] == 422 # a reason is required
changed = must(api("POST", "/products/DRAFT-1/change-product-type", CATALOGUE, {"product_type_code": "TRADING", "reason": "it is resold"}))
assert changed["product_type_code"] == "TRADING" and any("RAW → TRADING: it is resold" in h["detail"] for h in changed["history"])
sells("USED-1", "RAW")
must(api("POST", "/products/USED-1/activate-product", CATALOGUE, {}))
must(api("POST", "/products/USED-1/units/C62/used", SELLING, {"document_ref": "receipt R-1"}))
status, body = api("POST", "/products/USED-1/change-product-type", CATALOGUE, {"product_type_code": "TRADING", "reason": "oops"})
assert status == 409 and "was used by receipt R-1 for USED-1" in body["detail"]
assert api("POST", "/products/DRAFT-1/change-product-type", SALES, {"product_type_code": "RAW", "reason": "x"})[0] == 403
# the customer's own type, with its own code format
must(api("POST", "/product-types/SPARE", CATALOGUE, {"name": "Spare parts", "code_format": "SP-{seq:4}", "required_roles": ["purchase"]}))
assert sells(None, "SPARE")["product_id"] == "SP-0001"
def test_a_racing_add_loses_its_number_claim_and_takes_the_next(monkeypatch):
"""The series number is a plain-INSERT claim: an add that read a stale 'next' collides on (type, number) and is re-run on the
next free one (two concurrent adds get two numbers, never one twice)."""
from brick_product.infrastructure import tables
sells(None, "TRADING")
real, seen = tables.TypeRepository.next_number, []
def stale(self, *a):
seen.append(1)
return 1 if len(seen) == 1 else real(self, *a)
real_in_use = tables.ProductRepository.code_in_use
monkeypatch.setattr(tables.TypeRepository, "next_number", stale)
monkeypatch.setattr(tables.ProductRepository, "code_in_use", lambda self, code: None if len(seen) == 1 else real_in_use(self, code)) # the racer has not seen TRD-00001 yet
assert sells(None, "TRADING")["product_id"] == "TRD-00002" and len(seen) == 2
def screws():
for code, body in (("MATERIAL", {"data_type": "list", "allowed_values": ["A2", "A4", "ZINC"]}),
("LENGTH", {"data_type": "number", "unit_code": "MMT", "decimals": 1}),
("THREAD", {"data_type": "list", "allowed_values": ["M3", "M4", "M5"]})):
must(api("POST", f"/characteristics/{code}", CATALOGUE, body))
must(api("POST", "/product-classes/SCREWS", CATALOGUE, {"characteristics": [
{"characteristic_code": "MATERIAL", "required": True}, {"characteristic_code": "THREAD", "required": True},
{"characteristic_code": "LENGTH"}]}))
for pid in ("SCREW-M4-20", "SCREW-M4-30"):
must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": pid, "name": pid, "base_unit_code": "C62", "tax_category": "standard",
"billing_kind": "one_off", "invoicing_policy": "ordered"}))
def classify(pid, **kw):
return api("POST", f"/products/{pid}/classify-product", CATALOGUE, {"class_code": "SCREWS", **kw})
def search(*char, **kw):
return [r["product_id"] for r in must(api("GET", "/products", CATALOGUE, {"char": list(char), **kw}))["products"]]
def test_prd_36_products_are_found_by_characteristic_values_with_exact_unit_conversion():
"""PRD-36 — SCREWS requires MATERIAL and THREAD; values are dated rows; 2 CMT is 20 MMT; BRASS is refused; search converts exactly."""
screws()
must(classify("SCREW-M4-20", values=[{"characteristic_code": "MATERIAL", "value": "A2"}, {"characteristic_code": "LENGTH", "value": "20", "unit_code": "MMT"}]))
status, body = api("POST", "/products/SCREW-M4-20/activate-product", CATALOGUE, {})
assert status == 409 and "THREAD" in body["detail"] and "SCREWS" in body["detail"], body # class_complete
must(classify("SCREW-M4-20", values=[{"characteristic_code": "THREAD", "value": "M4"}]))
must(classify("SCREW-M4-30", values=[{"characteristic_code": "MATERIAL", "value": "A4"}, {"characteristic_code": "THREAD", "value": "M4"},
{"characteristic_code": "LENGTH", "value": "30 MMT"}]))
got = search("MATERIAL:=A2", "LENGTH:>=20 MMT")
assert got == ["SCREW-M4-20"]
assert search("LENGTH:>=20 MMT") == ["SCREW-M4-20", "SCREW-M4-30"] and search("THREAD:in M4,M5", class_code="SCREWS") == ["SCREW-M4-20", "SCREW-M4-30"]
assert search("LENGTH:<=2 CMT") == ["SCREW-M4-20"] # 2 CMT = 20 MMT, compared exactly
one = must(api("GET", "/products", CATALOGUE, {"char": ["MATERIAL:=A2"]}))["products"][0]
assert one["class_code"] == "SCREWS" and one["values"] == {"MATERIAL": "A2", "THREAD": "M4", "LENGTH": "20"}
# 2 CMT is stored as 20 MMT; BRASS is refused with the allowed list
with on_day("2026-11-21T09:00:00+00:00"):
changed = must(classify("SCREW-M4-20", values=[{"characteristic_code": "LENGTH", "value": "2", "unit_code": "CMT"}]))
assert changed["values"]["LENGTH"] == "20"
status, body = classify("SCREW-M4-20", values=[{"characteristic_code": "MATERIAL", "value": "BRASS"}])
assert status == 409 and "MATERIAL allows A2, A4, ZINC" in body["detail"]
assert classify("SCREW-M4-20", values=[{"characteristic_code": "LENGTH", "value": "2 KGM"}])[0] == 409 # a mass is not a length
assert classify("SCREW-M4-20", values=[{"characteristic_code": "LENGTH", "value": "1e15"}])[0] in (409, 422)
# a value change is a dated row: active products are reclassified from today, the past stays answerable
must(api("POST", "/products/SCREW-M4-20/activate-product", CATALOGUE, {}))
assert classify("SCREW-M4-20", values=[{"characteristic_code": "LENGTH", "value": "25"}])[0] == 409 # active: reclassify_product
with on_day("2026-11-22T09:00:00+00:00"):
must(api("POST", "/products/SCREW-M4-20/reclassify-product", CATALOGUE, {"values": [{"characteristic_code": "LENGTH", "value": "35"}]}))
assert search("LENGTH:>=30 MMT") == ["SCREW-M4-20", "SCREW-M4-30"] and search("LENGTH:>=30 MMT", on="2026-11-20") == ["SCREW-M4-30"]
assert search("LENGTH:>=30 MMT") == ["SCREW-M4-30"]
# permissions: only product:class:write classifies
assert api("POST", "/products/SCREW-M4-30/classify-product", SALES, {"class_code": "SCREWS"})[0] == 403
assert api("GET", "/products", SALES, {"char": ["MATERIAL:=A4"]})[0] == 200
def test_a_templates_values_are_inherited_and_a_template_attribute_names_its_characteristic():
"""PRD-36 — a template's values reach its variants unless one sets its own; generated variants carry the attribute's value."""
screws()
must(api("POST", "/characteristics/VOLUME", CATALOGUE, {"data_type": "number", "unit_code": "LTR", "decimals": 3}))
must(api("POST", "/product-classes/DRINKS", CATALOGUE, {"characteristics": [{"characteristic_code": "VOLUME", "required": True}]}))
made = must(api("POST", "/templates/COLA/generate-variants", CATALOGUE, {**BASE, "class_code": "DRINKS", "attributes": [
{**VOLUMES[0], "characteristic_code": "VOLUME"}]}))
assert made["created"] == ["COLA-033", "COLA-075", "COLA-150"]
assert must(api("GET", "/products/COLA-075", CATALOGUE))["characteristics"] == {"VOLUME": "0.75"}
assert search("VOLUME:=0.75 L", class_code="DRINKS") == ["COLA-075"] and search("VOLUME:>=500 MLT") == ["COLA-075", "COLA-150"]
must(api("POST", "/product-classes/BRAND", CATALOGUE, {"characteristics": [{"characteristic_code": "MATERIAL"}]}))
must(api("POST", "/products/COLA-075/classify-product", CATALOGUE, {"class_code": "BRAND", "scope": "template", "values": [{"characteristic_code": "MATERIAL", "value": "ZINC"}]}))
assert search("MATERIAL:=ZINC") == ["COLA-033", "COLA-075", "COLA-150"] # inherited by every variant
must(api("POST", "/products/COLA-150/classify-product", CATALOGUE, {"class_code": "BRAND", "values": [{"characteristic_code": "MATERIAL", "value": "A2"}]}))
assert search("MATERIAL:=ZINC") == ["COLA-033", "COLA-075"] and search("MATERIAL:=A2") == ["COLA-150"] # its own value wins
# ── site data, rounding profiles and the planning reads (PRD-37, PRD-38, PRD-39) ────────────────
import pytest # noqa: E402
from conftest import as_role # noqa: E402
from erp_platform import seam # noqa: E402
from fastapi import FastAPI, Response # noqa: E402
PLANNER, SYSTEM = as_role("planner"), as_role("system")
TOMORROW, MONDAY = "2026-11-21", "2026-11-23"
WAREHOUSES = {"WH1": "GB1", "WH2": "GB1", "WH3": "DE1"}
@pytest.fixture
def inventory():
"""A stand-in inventory that answers its signed GET /warehouses/{warehouse_id}: WH1 (Leeds) and WH2 (York) of company GB1, WH3 of DE1."""
app = FastAPI()
@app.get("/warehouses/{warehouse_id}")
def warehouse(warehouse_id: str, response: Response):
if warehouse_id not in WAREHOUSES:
response.status_code = 404
return {"detail": "no such warehouse"}
return {"warehouse_id": warehouse_id, "company_code": WAREHOUSES[warehouse_id], "status": "active"}
seam.register_local("inventory", app)
yield
seam.unregister_local("inventory")
def cola75(pid="COLA-075"):
"""COLA-075, base LTR: C62 = 3/4 LTR, PK24 = 18 LTR, bought in PK24, at least 10 PK24 in multiples of 10 from tomorrow."""
must(api("POST", "/products", CATALOGUE, {"product_type_code": "SERVICE", "product_id": pid, "name": "Cola 0.75 L", "base_unit_code": "LTR", "tax_category": "standard",
"billing_kind": "one_off", "invoicing_policy": "ordered"}))
must(api("POST", f"/products/{pid}/set-units", CATALOGUE, {"sales_unit_code": "C62", "purchase_unit_code": "PK24", "units_of": [
{"unit_code": "C62", "numerator": 3, "denominator": 4}, {"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24", "einvoice_unit_code": "XPK"}]}))
must(api("POST", f"/products/{pid}/activate-product", CATALOGUE, {}))
must(api("POST", f"/products/{pid}/set-rounding", CATALOGUE, {"role": "purchase", "unit_code": "PK24", "min_quantity": "10", "rounding_multiple": "10"}))
def profile(code="PALLET-LAYER"):
return must(api("POST", f"/rounding-profiles/{code}", CATALOGUE, {"unit_code": "PK24", "steps": [
{"multiple": "80", "round_up_pct": "90"}, {"multiple": "10", "round_up_pct": "50"}]}))
def site(site_id, **kw):
return api("POST", f"/products/COLA-075/sites/{site_id}", kw.pop("who", PLANNER), {"valid_from": TOMORROW, **kw})
def test_prd_37_site_data_is_a_dated_row_per_product_and_site(inventory):
"""PRD-37 — WH1 buys on min_max, WH2 is supplied from WH1 and is blocked for sale from Monday; each a dated row after today."""
cola75()
profile()
wh1 = must(site("WH1", procurement_type="buy", planned_delivery_days=5, gr_processing_days=1, planning_method="min_max", reorder_point="360",
safety_stock="90", max_stock="1440", rounding_profile_code="PALLET-LAYER"))
assert wh1["valid_from"] == TOMORROW and wh1["in_purchase_unit"]["reorder_point"] == "20" and wh1["in_purchase_unit"]["max_stock"] == "80"
must(site("WH2", procurement_type="transfer", supplying_site_id="WH1", planned_delivery_days=2, planning_method="reorder_point",
reorder_point="72", lot_size_method="fixed", fixed_lot_size="144"))
must(site("WH2", valid_from=MONDAY, site_status="no_sale")) # blocked for sale from next Monday; the rest carries forward
got = must(api("GET", "/products/COLA-075/sites/WH1", SALES, {"on": TOMORROW}))
assert (got["procurement_type"], got["planned_delivery_days"], got["gr_processing_days"], got["planning_method"], got["reorder_point"],
got["safety_stock"], got["max_stock"], got["rounding_profile_code"], got["base_unit_code"], got["purchase_unit_code"]) == \
("buy", 5, 1, "min_max", "360", "90", "1440", "PALLET-LAYER", "LTR", "PK24")
assert got["in_purchase_unit"] == {"reorder_point": "20", "safety_stock": "5", "max_stock": "80", "fixed_lot_size": None}
friday = must(api("GET", "/products/COLA-075/sites/WH2", SALES, {"on": "2026-11-22"}))
monday = must(api("GET", "/products/COLA-075/sites/WH2", SALES, {"on": MONDAY}))
assert friday["site_status"] == "active" and (monday["site_status"], monday["procurement_type"], monday["supplying_site_id"], monday["valid_from"]) == \
("no_sale", "transfer", "WH1", MONDAY)
assert (monday["fixed_lot_size"], monday["lot_size_method"], monday["in_purchase_unit"]["fixed_lot_size"]) == ("144", "fixed", "8")
today = must(api("GET", "/products/COLA-075/sites/WH2", SALES)) # before any row is in force, and at a site with none
for answer in (today, must(api("GET", "/products/COLA-075/sites/WH9", SALES))):
assert (answer["valid_from"], answer["site_status"], answer["procurement_type"], answer["planning_method"]) == (None, "active", "buy", "none")
# refused, with the reason
for kw, words in (({"planning_method": "min_max", "reorder_point": "2000", "max_stock": "1440"}, "reorder point (2000"),
({"planning_method": "reorder_point", "reorder_point": "50", "safety_stock": "60"}, "safety stock (60"),
({"procurement_type": "transfer", "supplying_site_id": "WH1"}, "cannot be supplied by itself"),
({"procurement_type": "transfer", "supplying_site_id": "WH3"}, "same company"),
({"valid_from": "2026-11-20"}, "after today"), ({"valid_from": "2026-11-01"}, "after today"),
({"procurement_type": "make"}, "need the production module"), ({"planning_method": "mrp"}, "need the production module")):
status, body = site("WH1", **kw)
assert status == 409 and words in body["detail"], (kw, status, body)
assert site("WH9", planning_method="manual")[0] == 422 and "no warehouse WH9" in site("WH9", planning_method="manual")[1]["detail"]
assert site("WH1", who=CATALOGUE, planning_method="manual")[0] == 403 and site("WH1", who=SALES, planning_method="manual")[0] == 403
assert api("GET", "/products/COLA-075/sites/WH1", SALES)[0] == 200 # read by everyone with product:product:read
seam.unregister_local("inventory") # without inventory the site is validated against nothing
assert site("WH-ANY", planning_method="manual")[0] == 200
def test_prd_38_a_proposal_rounds_to_the_pallet_or_layer_it_nearly_fills(inventory):
"""PRD-38 — pallet 80 from 90%, layer 10 from 50%: 75 -> 80, 47 -> 50, 163 -> 160; checked against the purchase rule; a person's quantity is not rounded."""
cola75()
profile()
must(site("WH1", planning_method="min_max", reorder_point="360", max_stock="1440", rounding_profile_code="PALLET-LAYER"))
must(site("WH2", planning_method="manual"))
def ask(q, site_id="WH1"):
return must(api("GET", "/products/COLA-075/round-proposal", SYSTEM, {"site_id": site_id, "quantity": q, "unit_code": "PK24", "on": TOMORROW}))
assert ask("75")["rounded"] == "80" and ask("75")["steps_applied"] == [{"multiple": "80", "round_up_pct": "90", "result": "80"}]
assert ask("47")["rounded"] == "50" and ask("47")["steps_applied"] == [{"multiple": "10", "round_up_pct": "50", "result": "50"}]
one63 = ask("163")
assert one63["rounded"] == "160" and one63["steps_applied"] == [{"multiple": "80", "round_up_pct": "90", "result": "160"}]
assert ask("163") == one63 and one63["purchase_rule"]["met"] is True # derived and repeatable
assert ask("3")["rounded"] == "10" # never rounded away: the smallest step, meeting the minimum
# a site with no profile: the proposal where the rule is met, else the rule's next allowed quantity
assert ask("80", "WH2")["rounded"] == "80" and ask("80", "WH2")["steps_applied"] == []
unmet = ask("75", "WH2")
assert unmet["rounded"] == "80" and unmet["purchase_rule"]["met"] is False and unmet["purchase_rule"]["adjusted_to"] == "80"
# a person's own quantity is gated by the purchase rule alone, never rounded by the profile
mine = must(api("GET", "/products/COLA-075/rounding", CATALOGUE, {"role": "purchase", "quantity": "75", "unit_code": "PK24", "on": TOMORROW}))
assert mine["met"] is False and mine["next_up"] == "80"
# a step that is not a multiple of the purchase rule's multiple is refused when the profile is saved
status, body = api("POST", "/rounding-profiles/PALLET-LAYER", CATALOGUE, {"unit_code": "PK24", "steps": [{"multiple": "85", "round_up_pct": "90"}, {"multiple": "15", "round_up_pct": "50"}]})
assert status == 409 and "not a multiple of 10 PK24" in body["detail"], body
must(api("POST", "/rounding-profiles/ODD", CATALOGUE, {"unit_code": "PK24", "steps": [{"multiple": "15", "round_up_pct": "50"}]})) # nothing uses it yet
status, body = site("WH2", rounding_profile_code="ODD")
assert status == 409 and "not a multiple of 10 PK24" in body["detail"]
assert api("POST", "/rounding-profiles/BAD", CATALOGUE, {"unit_code": "PK24", "steps": [{"multiple": "10", "round_up_pct": "50"}, {"multiple": "80", "round_up_pct": "90"}]})[0] == 409 # largest first
def test_prd_39_the_planning_reads_are_derived_from_the_dated_rows_and_repeat(inventory):
"""PRD-39 — inventory asks GET /product-sites for a site's planned products on its business date; sales-order asks one product at a site."""
cola75()
must(site("WH1", planning_method="min_max", reorder_point="360", max_stock="1440", planned_delivery_days=5))
must(site("WH2", procurement_type="transfer", supplying_site_id="WH1", planning_method="manual"))
cola75("COLA-150")
must(api("POST", "/products/COLA-150/sites/WH1", PLANNER, {"valid_from": TOMORROW, "planning_method": "none"})) # a row that plans nothing
ask = lambda **q: must(api("GET", "/product-sites", SYSTEM, {"site_id": "WH1", **q})) # noqa: E731
got = ask(planning="active", on=TOMORROW)
assert [(r["product_id"], r["valid_from"], r["planning_method"], r["reorder_point"], r["base_unit_code"], r["purchase_unit_code"]) for r in got] == \
[("COLA-075", TOMORROW, "min_max", "360", "LTR", "PK24")]
assert [r["product_id"] for r in ask(on=TOMORROW)] == ["COLA-075", "COLA-150"] and ask(planning="active", on=TOMORROW) == got
assert ask(planning="active") == [] # today: no row is in force yet
assert [r["product_id"] for r in ask(on=TOMORROW, after="COLA-075")] == ["COLA-150"] and len(ask(on=TOMORROW, limit=1)) == 1
two = lambda: must(api("GET", "/products/COLA-075/sites/WH2", SALES, {"on": "2026-12-02"})) # noqa: E731
assert two() == two() and two()["procurement_type"] == "transfer" and two()["planned_delivery_days"] is None
assert api("GET", "/product-sites", SALES, {"site_id": "WH1", "planning": "bogus"})[0] == 422