dodil/erp-brick-product
tests/test_audit6.py
"""The audit's round-5 follow-ups (erp-method 00daae3): a fixed rule keeps its unit; a used template row stays; the import refuses a
template id; no 500 on an overflow; a barcode race is a 409; a generate race is a 200; a used conversion freezes its level and
decimals; mark_unit_used is a machine permission; a pack code never shadows a UN/ECE code; a pack needs its e-invoice code."""
from conftest import (
CATALOGUE,
PRICING,
SALES,
SELLING,
SYSTEM,
api,
categories,
mk_list,
mk_product,
must,
person,
price,
row,
)
from sqlalchemy.exc import IntegrityError
from brick_product.domain import variants
def cola(pid="COLA-075"):
mk_product(pid, name="Cola", base_unit_code="LTR", activate=False)
must(api("POST", f"/products/{pid}/set-units", CATALOGUE, {"sales_unit_code": "C62", "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, {}))
def test_invariant_changing_a_sales_unit_changes_no_price_for_today_or_earlier_a_fixed_rule_keeps_its_unit():
"""Repro: a 1.10 fixed rule written per bottle (C62, then the sales unit); the pack becomes the sales unit; a past date still
answers 1.10 per bottle — the rule carries the unit it was written in and never looks at the product again."""
cola()
mk_list("LIST-GBP", default_from="2026-10-02", rules=[{"seq": 1, "applies_to": "product", "target": "COLA-075", "method": "fixed",
"fixed_price": "1.10"}])
assert must(api("GET", "/price-lists/LIST-GBP/rules", PRICING))["rules"][0]["unit_code"] == "C62" # set at write time
before = must(price("COLA-075", priced_on="2026-12-01"))
assert (before["unit_price"], before["unit_code"], before["rule"]["unit_code"]) == ("1.10", "C62", "C62")
must(api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24",
"einvoice_unit_code": "XPK"}, "use_for": ["sales"]}))
assert must(api("GET", "/products/COLA-075", CATALOGUE))["sales_unit_code"] == "PK24" # the sales unit changed …
after = must(price("COLA-075", unit_code="C62", priced_on="2026-12-01"))
assert (after["unit_price"], after["unit_code"]) == ("1.10", "C62") # … and a bottle still costs 1.10
pack = must(price("COLA-075", priced_on="2026-12-01")) # asked in the new sales unit
assert (pack["unit_code"], pack["unit_price"]) == ("PK24", "26.40") # 24 x 1.10, exactly
# a rule on a category or all needs the unit said outright: there is no one product whose sales unit it could be
mk_list("OTHER", publish=False)
status, body = api("POST", "/price-lists/OTHER/set-prices", PRICING, {"rules": [{"seq": 1, "method": "fixed", "fixed_price": "1.00"}]})
assert status == 409 and "stated per unit" in body["detail"]
mk_product("SVC", name="Service", base_unit_code="HUR")
must(api("POST", "/price-lists/OTHER/set-prices", PRICING, {"rules": [{"seq": 1, "method": "fixed", "fixed_price": "1.00", "unit_code": "C62"}],
"rows": [row("SVC", "2026-11-25", "50.00")]})) # a product not counted in C62: no match
def test_invariant_a_used_template_row_is_not_overridden_nor_are_its_level_or_decimals_changed():
"""Repro: template CS12 = 12 L, used by a variant's document; add_unit CS12 = 6 on another variant is a 409."""
categories(("drinks", None))
must(api("POST", "/templates/T/generate-variants", CATALOGUE, {"product_type_code": "LEGACY", "name": "T", "base_unit_code": "LTR", "category_code": "drinks", "tax_category": "standard",
"billing_kind": "one_off", "invoicing_policy": "ordered", "attributes": [{"name": "size", "values": [{"value": "a", "code": "1"}, {"value": "b", "code": "2"}]}],
"units_of": [{"unit_code": "CS12", "numerator": 12, "denominator": 1, "packaging_level": "case", "einvoice_unit_code": "XBX"}]}))
for sku in ("T-1", "T-2"):
must(api("POST", f"/products/{sku}/activate-product", CATALOGUE, {}))
assert must(api("GET", "/products/T-2/convert", CATALOGUE, {"quantity": "1", "from_unit": "CS12", "to_unit": "LTR"}))["converted"] == "12.000"
override = {"unit": {"unit_code": "CS12", "numerator": 6, "denominator": 1}}
must(api("POST", "/products/T-1/units/CS12/used", SELLING, {"document_ref": "SO-1"})) # the template row is used
for what in (override, {"unit": {"unit_code": "CS12", "numerator": 12, "denominator": 1, "packaging_level": "pallet"}},
{"unit": {"unit_code": "CS12", "numerator": 12, "denominator": 1, "quantity_decimals": 2}}):
status, body = api("POST", "/products/T-1/add-unit", CATALOGUE, what)
assert status == 409 and "already used" in body["detail"] and "through its template T" in body["detail"], body
assert api("POST", "/products/T-2/add-unit", CATALOGUE, what)[0] == 409 # the template row is used by T-1 too
assert must(api("POST", "/products/T-1/add-unit", CATALOGUE, {"unit": {"unit_code": "CS12", "numerator": 12, "denominator": 1}}))["units"]
assert must(api("GET", "/products/T-1/convert", CATALOGUE, {"quantity": "1", "from_unit": "CS12", "to_unit": "LTR"}))["converted"] == "12.000"
def test_a_used_pack_keeps_its_level_decimals_and_einvoice_code_too():
cola()
must(api("POST", "/products/COLA-075/units/PK24/used", SELLING, {}))
for change in ({"packaging_level": "pallet"}, {"quantity_decimals": 3}, {"einvoice_unit_code": "XPX"}):
status, body = api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "numerator": 18, "denominator": 1, **change}})
assert status == 409 and "NEW unit code" in body["detail"], change
assert api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "numerator": 18, "denominator": 1}})[0] == 200
def test_invariant_the_import_never_accepts_a_template_id_as_a_product():
categories(("drinks", None))
must(api("POST", "/templates/COLA/generate-variants", CATALOGUE, {"product_type_code": "LEGACY", "name": "Cola", "base_unit_code": "LTR", "attributes": [
{"name": "v", "values": [{"value": "x", "code": "1"}]}]}))
status, body = api("POST", "/catalogue/import", SYSTEM, {"products": [{"product_id": "COLA", "name": "Cola", "product_type_code": "LEGACY"}]})
assert status == 409 and "is a product template" in body["detail"]
assert api("GET", "/products/COLA", CATALOGUE)[0] == 404
assert must(api("POST", "/catalogue/import", SYSTEM, {"products": [{"product_id": "COLA-1", "name": "Cola x", "product_type_code": "LEGACY"}]}))["products_unchanged"] == 1
def test_a_huge_conversion_times_a_price_is_a_422_never_a_500():
mk_product("BIG", name="Big", base_unit_code="C62", activate=False)
must(api("POST", "/products/BIG/set-units", CATALOGUE, {"units_of": [{"unit_code": "BIGPK", "numerator": 2 ** 63 - 1, "denominator": 1,
"einvoice_unit_code": "XPK"}]}))
must(api("POST", "/products/BIG/activate-product", CATALOGUE, {}))
mk_list("LIST-GBP", default_from="2026-10-02", rows=[row("BIG", "2026-11-01", "99999999.9999", unit_code="C62")])
status, body = api("GET", "/products/BIG/price", SALES, {"currency": "GBP", "unit_code": "BIGPK", "priced_on": "2026-12-01"})
assert status == 422 and "cannot be stored" in body["detail"]
assert api("GET", "/products/BIG/price", SALES, {"currency": "GBP", "unit_code": "BIGPK", "quantity": "999999999999", "priced_on": "2026-12-01"})[0] == 422
assert must(price("BIG", priced_on="2026-12-01"))["unit_price"] == "99999999.9999"
assert must(api("GET", "/products/BIG/convert", CATALOGUE, {"quantity": "999999999999", "from_unit": "BIGPK", "to_unit": "C62"}))["converted"]
def test_a_barcode_claimed_by_someone_else_a_moment_ago_is_a_409_not_a_500(monkeypatch):
from brick_product.infrastructure import tables
cola()
def boom(self, *a, **k):
raise IntegrityError("insert", {}, Exception("duplicate key value violates unique constraint (SQLSTATE 23505)"))
monkeypatch.setattr(tables.GtinRepository, "claim", boom)
status, body = api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "numerator": 18, "denominator": 1,
"gtin": variants.with_check_digit("540000000009")}})
assert status == 409 and "claimed by someone else a moment ago" in body["detail"]
def test_two_generates_at_once_the_loser_finds_the_winners_variants_and_answers_200(monkeypatch):
"""A SKU that already belongs to THIS template with the same attribute values counts as existing — in the same attempt, or after a
retry — so the loser of the race gets 200."""
from brick_product.infrastructure import tables
one = {"product_type_code": "LEGACY", "name": "Cola", "base_unit_code": "LTR", "attributes": [{"name": "v", "values": [{"value": "a", "code": "1"}]}]}
both = {**one, "attributes": [{"name": "v", "values": [{"value": "a", "code": "1"}, {"value": "b", "code": "2"}]}]}
must(api("POST", "/templates/COLA/generate-variants", CATALOGUE, one)) # the winner has already committed COLA-1
state = {"stale": 2}
reads_products, reads_code = tables.TemplateRepository.products, tables.ProductRepository.code_in_use
def stale_products(self, template_id): # the loser read BEFORE the winner committed
if state["stale"]:
return []
return reads_products(self, template_id)
def stale_code(self, code):
if state["stale"]:
return None
return reads_code(self, code)
monkeypatch.setattr(tables.TemplateRepository, "products", stale_products)
monkeypatch.setattr(tables.ProductRepository, "code_in_use", stale_code)
real_save = tables.ProductRepository.save
def save(self, agg, expected):
try:
return real_save(self, agg, expected)
finally:
state["stale"] = 0 # its retry reads fresh
monkeypatch.setattr(tables.ProductRepository, "save", save)
status, body = api("POST", "/templates/COLA/generate-variants", CATALOGUE, both)
assert status == 200, (status, body)
assert body["existing"] == ["COLA-1"] and body["created"] == ["COLA-2"]
assert len(must(api("GET", "/products", CATALOGUE))["products"]) == 2
def test_mark_unit_used_is_a_machine_permission_held_by_a_selling_service_alone():
"""No person holds it, whatever their wildcard; the selling service and the system (signed spec v4 actor `system`) do."""
cola()
for who in (SALES, PRICING, CATALOGUE, person("x@shop.example", ["*:*:*"]), person("y@shop.example", ["product:*:*"])):
assert api("POST", "/products/COLA-075/units/C62/used", who, {})[0] == 403
assert must(api("POST", "/products/COLA-075/units/C62/used", SELLING, {}))["used"] is True
assert must(api("POST", "/products/COLA-075/units/C62/used", SYSTEM, {}))["used"] is True
def test_invariant_a_pack_needs_its_einvoice_unit_code_before_a_document_may_use_it():
mk_product("COLA-075", name="Cola", base_unit_code="LTR", activate=False)
must(api("POST", "/products/COLA-075/set-units", CATALOGUE, {"units_of": [
{"unit_code": "PK24", "numerator": 18, "denominator": 1}, {"unit_code": "XBX", "numerator": 6, "denominator": 1}]}))
must(api("POST", "/products/COLA-075/activate-product", CATALOGUE, {}))
status, body = api("POST", "/products/COLA-075/units/PK24/used", SELLING, {})
assert status == 409 and "no e-invoice unit code yet" in body["detail"] and "XPK" in body["detail"]
assert must(api("POST", "/products/COLA-075/units/XBX/used", SELLING, {}))["used"] # a UN/ECE code is its own
assert must(api("POST", "/products/COLA-075/units/LTR/used", SELLING, {}))["used"] # so is the base unit
assert api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "numerator": 18, "denominator": 1, "einvoice_unit_code": "x"}})[0] in (409, 422)
must(api("POST", "/products/COLA-075/add-unit", CATALOGUE, {"unit": {"unit_code": "PK24", "numerator": 18, "denominator": 1, "einvoice_unit_code": "XPK"}}))
assert must(api("POST", "/products/COLA-075/units/PK24/used", SELLING, {}))["used"]
mk_list("LIST-GBP", default_from="2026-10-02", rows=[row("COLA-075", "2026-11-01", "20.00", unit_code="PK24")])
got = must(price("COLA-075", unit_code="PK24", priced_on="2026-12-01"))
assert got["conversion"]["einvoice_unit_code"] == "XPK" and got["line"]["einvoice_unit_code"] == "XPK"
def test_a_pack_code_never_shadows_a_un_ece_code_unless_it_carries_that_codes_real_conversion():
mk_product("PCS", name="Pieces", base_unit_code="C62", activate=False)
status, body = api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": "KGM", "numerator": 25, "denominator": 1}]})
assert status == 409 and "KGM is the UN/ECE code for a unit of mass" in body["detail"] and "PK25" in body["detail"] # KGM as 25 pieces
for code in ("LTR", "HUR", "MTR", "WEE"):
assert api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": code, "numerator": 2, "denominator": 1}]})[0] == 409
must(api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK25", "numerator": 25, "denominator": 1}, {"unit_code": "XBX", "numerator": 12, "denominator": 1}]}))
mk_product("SAND", name="Sand", base_unit_code="GRM", activate=False)
status, body = api("POST", "/products/SAND/set-units", CATALOGUE, {"units_of": [{"unit_code": "KGM", "numerator": 1000, "denominator": 1}]})
assert status == 409 and "the same way for every product" in body["detail"]
assert must(api("GET", "/products/SAND/convert", CATALOGUE, {"quantity": "2", "from_unit": "KGM", "to_unit": "GRM"}))["converted"] == "2000"
def test_the_whole_un_ece_measuring_set_is_reserved_not_only_the_seeded_codes():
mk_product("PCS", name="Pieces", base_unit_code="C62", activate=False)
for code in ("BAR", "KWH", "MTQ", "CMT", "FAH", "MMT", "NEW", "TNE", "LBR", "GLL"): # pressure, energy, volume, length …
status, body = api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": code, "numerator": 12, "denominator": 1}]})
assert status == 409 and "measuring unit" in body["detail"] and "your own pack code" in body["detail"], code
for code in ("H87", "EA", "SET", "PR", "DZN", "XBX", "XPX", "PK12", "PALLET80"): # counting, Rec 21 packaging, your own
assert api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": code, "numerator": 12, "denominator": 1}]})[0] == 200, code
from brick_product.domain import unece
assert len(unece.CODES) > 2000 and "XPK" in unece.CODES and len(unece.RESERVED) > 1500
def test_an_einvoice_unit_code_must_be_a_real_rec_20_or_21_code():
mk_product("PCS", name="Pieces", base_unit_code="C62", activate=False)
for bad in ("ZZZ", "AAA", "99", "ZZ", "xpk"):
assert api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "numerator": 6, "denominator": 1,
"einvoice_unit_code": bad}]})[0] in (409, 422), bad
status, body = api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "numerator": 6, "denominator": 1, "einvoice_unit_code": "ZZZ"}]})
assert status == 422 and "not a UN/ECE Recommendation 20 or 21 code" in body["detail"]
for good in ("XPK", "XPX", "XBX", "KGM", "C62"):
assert api("POST", "/products/PCS/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK6", "numerator": 6, "denominator": 1, "einvoice_unit_code": good}]})[0] == 200
def test_invariant_a_pack_without_an_einvoice_code_copies_none_onto_a_line_never_its_own_code():
mk_product("COLA-075", name="Cola", base_unit_code="LTR", activate=False)
must(api("POST", "/products/COLA-075/set-units", CATALOGUE, {"units_of": [{"unit_code": "PK24", "numerator": 18, "denominator": 1},
{"unit_code": "PK12", "numerator": 9, "denominator": 1, "einvoice_unit_code": "XPK"}]}))
must(api("POST", "/products/COLA-075/activate-product", CATALOGUE, {}))
mk_list("LIST-GBP", default_from="2026-10-02", rows=[row("COLA-075", "2026-11-01", "20.00", unit_code="PK24"),
row("COLA-075", "2026-11-01", "10.00", unit_code="PK12"), row("COLA-075", "2026-11-01", "1.00", unit_code="LTR")])
bare = must(price("COLA-075", unit_code="PK24", priced_on="2026-12-01"))
assert bare["conversion"]["einvoice_unit_code"] is None and bare["line"]["einvoice_unit_code"] is None
assert must(price("COLA-075", unit_code="PK12", priced_on="2026-12-01"))["line"]["einvoice_unit_code"] == "XPK"
assert must(price("COLA-075", unit_code="LTR", priced_on="2026-12-01"))["line"]["einvoice_unit_code"] == "LTR" # the base unit is its own
def test_a_rules_unit_code_must_be_a_known_unit():
cola()
mk_list("OTHER", publish=False)
for unit in ("NOPE", "KGM2", "ZZZ9"):
status, body = api("POST", "/price-lists/OTHER/set-prices", PRICING, {"rules": [{"seq": 1, "method": "fixed", "fixed_price": "1.00", "unit_code": unit}]})
assert status == 422 and "not a known unit" in body["detail"], unit
for unit in ("C62", "PK24", "KGM"): # a UN/ECE unit, or a pack some product has
assert api("POST", "/price-lists/OTHER/set-prices", PRICING, {"rules": [{"seq": 1, "method": "fixed", "fixed_price": "1.00", "unit_code": unit}]})[0] == 200
status, body = api("POST", "/price-lists/OTHER/set-prices", PRICING, {"rules": [{"seq": 2, "applies_to": "product", "target": "COLA-075", "method": "fixed",
"fixed_price": "1.00", "unit_code": "KGM"}]})
assert status == 422 and "COLA-075 is not counted in KGM" in body["detail"]