"""product's scenario catalogue (module.yaml tests.scenarios, SC-PRD-01..03), each mapped to its stories and written
ONCE over the composition's doors — routes and MCP tools — never a table. Each makes its own products and lists
(s.key), so none depends on another; each sets the business clock, so none depends on the day it runs.
erp-scenarios list tests/scenarios/catalogue.py
erp-scenarios run tests/scenarios/catalogue.py
The spec's SC-PRD-01 says the 10% rule comes first and the 15% break second; the first rule that matches wins, so the
break is the list's seq 1 here (the 10% would otherwise match 12 units first and the break would never fire).
"""
import pathlib
import uuid
from decimal import Decimal
from erp_platform.scenarios import Catalogue
from brick_product.domain.variants import with_check_digit
HERE = pathlib.Path(__file__).resolve().parent
ROLES = HERE / "roles.yaml" # the brick's default roles, as shipped, plus whoever keeps the partner register
CAT = Catalogue("product", compose=HERE / "compose.yaml", roles=ROLES, zone="UTC")
# Generated over the composition's doors, both HTTP and MCP (erp_platform.scenarios.probes): every role × every use
# case — 401 with no identity, 403 without the permission, served with it — and the wildcard probes, including
# PROBE-WILD: `*:*:*` and `product:*:*` run every person use case and no machine one, `zz:*:*` and malformed grants none.
CAT.permission_matrix()
SETUP = "2026-10-15T09:00:00Z" # lists are built a month earlier: nothing published may start on or before the day it is published
TODAY = "2026-11-20T09:00:00Z"
DEFAULT = "2026-10-16" # the day after SETUP
# Each scenario prices in its own currency: a currency has ONE current default list, and a new default may not
# back-date over the old (the scenarios share one bucket, and run in any order).
def new_product(s, cm, code, category="hosting", **kw):
"""The catalogue manager adds a product and offers it to sales."""
cm.post("/product/products", {"product_id": code, "product_type_code": "SERVICE", "name": f"Managed hosting {code}", "base_unit_code": "MON",
"category_code": category, "tax_category": "standard", "billing_kind": "recurring",
"invoicing_policy": "ordered", **kw}).ok()
cm.post(f"/product/products/{code}/activate-product", {}).ok().field("status", "active")
def new_list(pm, list_id, rows=(), rules=(), publish=True, currency="GBP", **kw):
pm.post("/product/price-lists", {"price_list_id": list_id, "currency": currency, **kw}).ok()
if rows or rules:
pm.post(f"/product/price-lists/{list_id}/set-prices",
{"rows": list(rows), "rules": [{"valid_from": DEFAULT, **r} for r in rules]}).ok()
if publish:
pm.post(f"/product/price-lists/{list_id}/publish-price-list", {}).ok().field("status", "published")
@CAT.scenario("SC-PRD-01", "a customer list with a quantity break", stories=["PRD-05", "PRD-06", "PRD-08", "PRD-11"],
tags=["money", "clock"], personas=["catalogue_manager", "pricing_manager", "sales_rep", "partner_steward"])
def sc_prd_01(s):
cm, pm, rep = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("sales_rep")
s.clock.set(SETUP)
host, standard, partner = s.key("HOST"), s.key("LIST"), s.key("PARTNER")
s.given("LIST-GBP HOST-M 120.00; PARTNER-GBP 10% off LIST-GBP for hosting, 15% off from 10 units; customer on PARTNER-GBP")
cm.post("/product/product-categories/hosting", {"name": "Hosting"}).ok()
new_product(s, cm, host)
new_list(pm, standard, rows=[{"product_id": host, "valid_from": "2026-11-01", "unit_price": "120.00"}],
default_from=DEFAULT)
new_list(pm, partner, rules=[
{"seq": 1, "applies_to": "category", "target": "hosting", "min_quantity": "10", "discount_pct": "15",
"base_list": standard, "valid_from": "2026-11-01"},
{"seq": 2, "applies_to": "category", "target": "hosting", "discount_pct": "10", "base_list": standard,
"valid_from": "2026-11-01"}])
customer = s.as_("partner_steward").put("/business-partner/partners", {"bp_key": s.key("CUST"), "legal_name": "Northwind Ltd",
"country": "GB", "roles": ["customer"]}).ok().value("bp_id")
pm.post(f"/product/price-lists/{partner}/assign-customer", {"bp_id": "9999999999"}).refused(422, "no business partner")
pm.post(f"/product/price-lists/{partner}/assign-customer", {"bp_id": customer, "valid_from": "2026-11-01"}).ok() \
.field("price_list_id", partner)
s.clock.set(TODAY)
ask = {"currency": "GBP", "priced_on": "2026-12-01"}
s.when("prices are asked for 5 and for 12 units for the customer, and for 12 units for a customer with no list")
five = rep.get(f"/product/products/{host}/price", quantity="5", bp_id=customer, **ask)
twelve = rep.get(f"/product/products/{host}/price", quantity="12", bp_id=customer, **ask)
stranger = rep.get(f"/product/products/{host}/price", quantity="12", bp_id="9999", **ask)
s.then("108.00, 102.00 and 120.00, each naming the list and rule used")
five.ok().field("unit_price", Decimal("108.00")).field("price_list_id", partner).field("rule.seq", 2)
twelve.ok().field("unit_price", Decimal("102.00")).field("price_list_id", partner).field("rule.seq", 1) \
.field("rule.method", "discount_pct").field("how.base.price_list_id", standard)
stranger.ok().field("unit_price", Decimal("120.00")).field("price_list_id", standard).field("via", "row")
stranger.field("price_row.valid_from", "2026-11-01").field("rule", None)
rep.tool("find_price", product_id=host, quantity="12", bp_id=customer, **ask).ok().field("unit_price", Decimal("102.00"))
twelve.same_as(rep.get(f"/product/products/{host}/price", quantity="12", bp_id=customer, **ask))
rep.get(f"/product/products/{host}/price", quantity="12", currency="NOK").not_found() # never converted: nobody has a NOK list
s.note("a rep never sees the standard cost, and cannot write a price")
rep.post(f"/product/price-lists/{standard}/set-prices", {}).forbidden("product:price_list:write")
@CAT.scenario("SC-PRD-02", "the annual increase", stories=["PRD-09", "PRD-07", "PRD-11"], tags=["money", "clock"],
personas=["catalogue_manager", "pricing_manager", "pricing_manager#2", "owner"])
def sc_prd_02(s):
cm, pm, pm2 = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("pricing_manager", 2)
owner = s.as_("owner")
s.clock.set(SETUP)
cat, standard = s.key("fam"), s.key("LIST")
s.given("40 hosting products on LIST-GBP")
cm.post(f"/product/product-categories/{cat}", {}).ok()
codes = [s.key(f"H{i:02d}") for i in range(40)]
for code in codes:
new_product(s, cm, code, category=cat)
new_list(pm, standard, rows=[{"product_id": c, "valid_from": "2026-11-01", "unit_price": "120.00"} for c in codes],
currency="EUR", default_from=DEFAULT, rounding="0.05")
s.clock.set(TODAY)
owner.post("/product/pricing-controls/publish_needs_second_person", {"value": True}).ok()
raise_ = {"raise_pct": "5", "valid_from": "2027-01-01", "category_code": cat}
s.when("a 5% rise from 2027-01-01 is drafted, published, and the draft run again")
pm.post(f"/product/price-lists/{standard}/change-prices", raise_).ok().field("written.rows", 40) \
.field("contents.rows_pending", 40)
pm.post(f"/product/price-lists/{standard}/publish-price-list", {}).refused(409, "second person") # the author may not
pm2.post(f"/product/price-lists/{standard}/publish-price-list", {}).ok().field("contents.rows_pending", 0)
pm.post(f"/product/price-lists/{standard}/change-prices", raise_).ok().field("written.rows", 0) \
.field("written.rows_unchanged", 40)
s.then("40 new rows; prices before 2027-01-01 unchanged; the re-run adds nothing")
pm.get(f"/product/price-lists/{standard}/rows", status="live", limit=1000).ok().length("rows", 80)
for code in (codes[0], codes[39]):
pm.get(f"/product/products/{code}/price", currency="EUR", priced_on="2026-12-31").ok().field("unit_price", Decimal("120.00"))
pm.get(f"/product/products/{code}/price", currency="EUR", priced_on="2027-01-01").ok().field("unit_price", Decimal("126.00"))
pm.post(f"/product/price-lists/{standard}/publish-price-list", {}).ok().same_as(
pm2.post(f"/product/price-lists/{standard}/publish-price-list", {}), ignore=()) # nothing waiting: a repeat changes nothing
owner.post("/product/pricing-controls/publish_needs_second_person", {"value": None}).ok()
@CAT.scenario("SC-PRD-03", "archive without breaking documents", stories=["PRD-03", "PRD-21"], tags=["clock"],
personas=["catalogue_manager", "pricing_manager", "sales_rep"])
def sc_prd_03(s):
cm, pm, rep = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("sales_rep")
s.clock.set(SETUP)
host, standard = s.key("HOST"), s.key("LIST")
cm.post("/product/product-categories/hosting", {}).ok()
new_product(s, cm, host)
new_list(pm, standard, rows=[{"product_id": host, "valid_from": "2026-11-01", "unit_price": "120.00"}],
currency="USD", default_from=DEFAULT)
s.clock.set(TODAY)
s.given("HOST-M on issued invoices and an open quote (each holds the copy a line took from find_price)")
copy = rep.get(f"/product/products/{host}/price", currency="USD", bp_id="7311").ok().field("unit_price", Decimal("120.00")).value("line")
s.when("it is archived")
cm.post(f"/product/products/{host}/archive-product", {}).ok().field("status", "archived")
s.then("a new line for HOST-M is refused; the invoices and the quote read exactly as before")
rep.get(f"/product/products/{host}/price", currency="USD", bp_id="7311").refused(409, "not offered for new lines")
rep.tool("find_price", product_id=host, currency="USD").refused(409, "archived")
rep.get(f"/product/products/{host}").not_found() # sales no longer sees it
held = cm.get(f"/product/products/{host}").ok().field("status", "archived")
for field in ("name", "description", "base_unit_code", "tax_category", "billing_kind", "invoicing_policy"):
held.field(field, copy[field]) # what the documents copied is still true
cm.post(f"/product/products/{host}/reactivate-product", {"reason": "back on the winter range"}).ok().field("status", "active")
rep.get(f"/product/products/{host}/price", currency="USD").ok().field("line", copy)
@CAT.scenario("SC-PRD-04", "a new default list never re-prices the past", stories=["PRD-05", "PRD-07", "PRD-08", "PRD-11"],
tags=["money", "clock"], personas=["catalogue_manager", "pricing_manager", "sales_rep"])
def sc_prd_04(s):
cm, pm, rep = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("sales_rep")
host, old, new, today_default = s.key("HOST"), s.key("LIST"), s.key("NEW"), s.key("TODAY")
s.clock.set("2025-12-15T09:00:00Z")
cm.post("/product/product-categories/hosting", {}).ok()
new_product(s, cm, host)
s.given("LIST-GBP the GBP default since 2026-01-01, HOST-M 100.00; customer 7311 with no assignment")
new_list(pm, old, rows=[{"product_id": host, "valid_from": "2026-01-01", "unit_price": "100.00"}], currency="SEK",
default_from="2026-01-01")
s.clock.set(TODAY)
ask = {"currency": "SEK", "bp_id": "7311"}
held = {d: rep.get(f"/product/products/{host}/price", priced_on=d, **ask).ok().field("unit_price", Decimal("100.00")).field("price_list_id", old)
for d in ("2026-01-01", "2026-06-30", "2026-11-19", "2026-11-20")}
s.when("NEW-GBP, based on LIST-GBP at 30% off, is published as the GBP default from tomorrow; then retiring LIST-GBP, a "
"default_from dated today and an assignment of 7311 from yesterday are tried")
pm.post("/product/price-lists", {"price_list_id": new, "currency": "SEK", "based_on": old, "default_from": "2026-11-21"}).ok()
pm.post(f"/product/price-lists/{new}/set-prices", {"rules": [
{"seq": 1, "discount_pct": "30", "valid_from": "2026-11-21", "base_list": old}]}).ok()
pm.post(f"/product/price-lists/{new}/publish-price-list", {}).ok().field("status", "published")
pm.post(f"/product/price-lists/{old}/retire-price-list", {}).refused(409, "default SEK list")
pm.post("/product/price-lists", {"price_list_id": today_default, "currency": "SEK", "default_from": "2026-11-20"}).ok()
pm.post(f"/product/price-lists/{today_default}/publish-price-list", {}).refused(409, "not after today")
pm.post(f"/product/price-lists/{new}/assign-customer", {"bp_id": "7311", "valid_from": "2026-11-19"}).refused(409, "tomorrow or later")
s.then("prices for today and earlier are still 100.00 from LIST-GBP and 70.00 from tomorrow; the retire, the dated-today "
"default and the back-dated assignment are refused")
for d, was in held.items():
rep.get(f"/product/products/{host}/price", priced_on=d, **ask).ok().same_as(was)
rep.get(f"/product/products/{host}/price", priced_on="2026-11-21", **ask).ok().field("unit_price", Decimal("70.00")) \
.field("price_list_id", new)
@CAT.scenario("SC-PRD-05", "a late publish never back-dates a price", stories=["PRD-07", "PRD-09", "PRD-11"],
tags=["money", "clock"], personas=["catalogue_manager", "pricing_manager", "sales_rep"])
def sc_prd_05(s):
"""Not in the spec's three: the audit's business outcome. A price change scheduled for 12-10 and published on 12-20
would answer 150.00 for a date on which documents were already priced at 120.00 — so it is refused, and re-dated."""
cm, pm, rep = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("sales_rep")
s.clock.set(SETUP)
host, standard = s.key("HOST"), s.key("LIST")
cm.post("/product/product-categories/hosting", {}).ok()
new_product(s, cm, host)
new_list(pm, standard, rows=[{"product_id": host, "valid_from": "2026-11-01", "unit_price": "120.00"}],
currency="CHF", default_from=DEFAULT)
s.clock.set(TODAY)
s.given("HOST-M at 120.00; a rise to 150.00 scheduled from 2026-12-10, not yet published")
pm.post(f"/product/price-lists/{standard}/change-prices", {"rows": [
{"product_id": host, "valid_from": "2026-12-10", "unit_price": "150.00"}]}).ok().field("contents.rows_pending", 1)
quoted = rep.get(f"/product/products/{host}/price", currency="CHF", priced_on="2026-12-10").ok().field("unit_price", Decimal("120.00"))
s.when("the publisher is late: it is 2026-12-20")
s.clock.set("2026-12-20T09:00:00Z")
pm.post(f"/product/price-lists/{standard}/publish-price-list", {}).refused(409, "already past")
s.then("the 2026-12-10 answer is still 120.00; the change is discarded and scheduled from a later date")
rep.get(f"/product/products/{host}/price", currency="CHF", priced_on="2026-12-10").ok().same_as(quoted)
pm.post(f"/product/price-lists/{standard}/change-prices", {
"discard": [{"product_id": host, "valid_from": "2026-12-10"}],
"rows": [{"product_id": host, "valid_from": "2027-01-01", "unit_price": "150.00"}]}).ok().field("written.discarded", 1)
pm.post(f"/product/price-lists/{standard}/publish-price-list", {}).ok().field("contents.rows_pending", 0)
rep.get(f"/product/products/{host}/price", currency="CHF", priced_on="2026-12-10").ok().same_as(quoted)
rep.get(f"/product/products/{host}/price", currency="CHF", priced_on="2027-01-01").ok().field("unit_price", Decimal("150.00"))
@CAT.scenario("SC-PRD-06", "a drink bought in packs, sold by the bottle, used in litres", stories=["PRD-23", "PRD-24", "PRD-25", "PRD-26"],
tags=["money", "clock"], personas=["catalogue_manager", "pricing_manager", "sales_rep", "selling_service"])
def sc_prd_06(s):
cm, pm, rep = s.as_("catalogue_manager"), s.as_("pricing_manager"), s.as_("sales_rep")
s.clock.set(SETUP)
cola, standard = s.key("COLA"), s.key("LIST")
sku = {v: f"{cola}-{c}" for v, c in (("0.33 L", "033"), ("0.75 L", "075"), ("1.5 L", "150"))}
bottle_sku = sku["0.75 L"]
digits = "".join(str(int(c, 16) % 10) for c in uuid.uuid4().hex)
pack_gtin, pallet_gtin = with_check_digit("54" + digits[:10]), with_check_digit("55" + digits[10:20])
s.given("template COLA (volume 0.33 L, 0.75 L, 1.5 L), base unit LTR; COLA-075 bottle = 0.75 LTR, PK24 = 24 bottles, PL = 80 PK24 "
"with GTINs; LIST-GBP 1.20 a bottle, 24.00 a PK24")
cm.post("/product/product-categories/drinks", {"name": "Drinks"}).ok()
template = {"product_type_code": "LEGACY", "name": "Cola", "base_unit_code": "LTR", "category_code": "drinks", "tax_category": "standard",
"billing_kind": "one_off", "invoicing_policy": "ordered",
"attributes": [{"name": "volume", "values": [{"value": v, "code": c} for v, c in
(("0.33 L", "033"), ("0.75 L", "075"), ("1.5 L", "150"))]}]}
s.when("the variants are generated twice, 2 PK24 and 48 bottles of COLA-075 are priced, the pallet GTIN is scanned, and the PK24 "
"conversion is changed after an order used it")
first = cm.post(f"/product/templates/{cola}/generate-variants", template).ok().field("created", sorted(sku.values(), key=str))
cm.post(f"/product/templates/{cola}/generate-variants", template).ok().field("created", []).length("existing", 3)
cm.get("/product/products", q=cola).ok().length("products", 3)
cm.post(f"/product/products/{bottle_sku}/set-units", {
"sales_unit_code": "C62", "purchase_unit_code": "PK24", "issue_unit_code": "LTR", "units_of": [
{"unit_code": "C62", "numerator": 3, "denominator": 4, "packaging_level": "each"},
{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24", "packaging_level": "case", "gtin": pack_gtin,
"einvoice_unit_code": "XPK"},
{"unit_code": "PALLET80", "contains_unit": "PK24", "contains_quantity": "80", "packaging_level": "pallet", "gtin": pallet_gtin,
"einvoice_unit_code": "XPX"}]}).ok()
cm.post(f"/product/products/{bottle_sku}/activate-product", {}).ok().field("status", "active")
new_list(pm, standard, currency="DKK", default_from=DEFAULT, rows=[
{"product_id": bottle_sku, "valid_from": "2026-11-01", "unit_price": "1.20", "unit_code": "C62"},
{"product_id": bottle_sku, "valid_from": "2026-11-01", "unit_price": "24.00", "unit_code": "PK24"}])
s.clock.set(TODAY)
ask = {"currency": "DKK", "priced_on": "2026-12-01"}
packs = rep.get(f"/product/products/{bottle_sku}/price", quantity="2", unit_code="PK24", **ask).ok().field("unit_code", "PK24")
bottles = rep.get(f"/product/products/{bottle_sku}/price", quantity="48", **ask).ok().field("unit_code", "C62")
scan = rep.get(f"/product/gtin/{pallet_gtin}").ok().field("product_id", bottle_sku).field("unit_code", "PALLET80")
s.then("three products, not six; 48.00 and 57.60; the pallet is 1 920 bottles (1 440 LTR); the changed conversion is refused and a "
"new pack unit is offered")
first.length("products", 3)
packs.matches("unit_price", lambda v: Decimal(v) * 2 == Decimal("48.00"), "2 packs cost 48.00")
bottles.matches("unit_price", lambda v: Decimal(v) * 48 == Decimal("57.60"), "48 bottles cost 57.60")
scan.matches("quantity_in_sales_unit", lambda v: Decimal(v) == 1920, "1 920 bottles").matches(
"quantity_in_base_unit", lambda v: Decimal(v) == 1440, "1 440 LTR")
rep.post(f"/product/products/{bottle_sku}/units/PK24/used", {"document_ref": "SO-1"}).forbidden("product:unit:mark_used") # a machine's
s.as_("selling_service").post(f"/product/products/{bottle_sku}/units/PK24/used", {"document_ref": "SO-1"}).ok() # an order used the pack
cm.post(f"/product/products/{bottle_sku}/add-unit", {"unit": {"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "12"}}) \
.refused(409, "NEW unit code")
cm.post(f"/product/products/{bottle_sku}/add-unit", {"unit": {"unit_code": "PK12", "contains_unit": "C62", "contains_quantity": "12"},
"use_for": ["purchase"]}).ok().field("purchase_unit_code", "PK12")
rep.get(f"/product/products/{bottle_sku}/price", quantity="2", unit_code="PK24", **ask).ok().same_as(packs) # the old pack: unchanged