dodil/erp-brick-product
tests/test_audit_v4.py
"""The v4 audit (slices A and B, PRD-27..39): each MUST-fix defect proven by a strict xfail — it fails today and turns the suite
red the day it is fixed, so the fix also removes the marker. Races are replayed in-process: the loser's read is taken, the winner
runs to commit in between, the loser goes on from what it read (the DataK3 pg wire has no row lock to stop it)."""
from contextlib import contextmanager
import pytest
from conftest import (
CATALOGUE,
DEFAULT,
SALES,
SELLING,
api,
as_role,
mk_list,
mk_product,
must,
person,
price,
row,
)
from sqlalchemy import event
from brick_product.infrastructure import tables
PLANNER = as_role("planner")
IMPORTER = person("importer@shop.example", ["product:catalogue:import", "product:migration:read", "product:product:read"])
MAIN, CASE = "05012345000015", "04006381333931"
def cola(pid="COLA-075", activate=True):
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"}]}))
if activate:
must(api("POST", f"/products/{pid}/activate-product", CATALOGUE, {}))
def winner_runs_between(monkeypatch, cls, method, winner):
"""Patch `cls.method` so its FIRST call (the loser's read) answers what it read BEFORE `winner()` ran and committed."""
real, state = getattr(cls, method), {"done": False}
def stale(self, *a, **k):
before = real(self, *a, **k)
if not state["done"]:
state["done"] = True
winner()
return before
monkeypatch.setattr(cls, method, stale)
return state
@contextmanager
def statements(engine):
seen = []
def count(conn, cursor, statement, params, context, executemany):
seen.append(statement)
event.listen(engine, "before_cursor_execute", count)
try:
yield seen
finally:
event.remove(engine, "before_cursor_execute", count)
# ── MUST-1: two operations on one product that share no claim key ─────────────────────────────
def test_must1_two_first_gtins_of_one_unit_racing_leave_exactly_one_main(monkeypatch):
"""PRD-27 'a unit has exactly one main GTIN while it has any'. Both adds read the unit's GTINs (none), both claim theirs as main:
the claims are different rows (key gtin), so nothing collides."""
cola()
real = tables.GtinRepository.of_unit
winner_runs_between(monkeypatch, tables.GtinRepository, "of_unit",
lambda: must(api("POST", "/products/COLA-075/add-gtin", CATALOGUE, {"unit_code": "PK24", "gtin": MAIN})))
api("POST", "/products/COLA-075/add-gtin", CATALOGUE, {"unit_code": "PK24", "gtin": CASE})
monkeypatch.setattr(tables.GtinRepository, "of_unit", real)
kinds = sorted(must(api("GET", f"/gtin/{g}", CATALOGUE))["kind"] for g in (MAIN, CASE))
assert kinds == ["additional", "main"], kinds
def screws():
for code, body in (("MATERIAL", {"data_type": "list", "allowed_values": ["A2", "A4", "ZINC"]}),
("THREAD", {"data_type": "list", "allowed_values": ["M3", "M4", "M5"]}),
("LENGTH", {"data_type": "number", "unit_code": "MMT", "decimals": 1})):
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"}]}))
def test_must1_a_classify_racing_the_activation_cannot_leave_an_incomplete_active_product(monkeypatch):
"""PRD-36 gate class_complete. activate reads the classification (none), classify (draft-only) puts the draft in SCREWS without
THREAD and commits, activate writes prod_product: an ACTIVE product in SCREWS with no THREAD."""
screws()
mk_product("SCREW-1", name="Screw", base_unit_code="C62", billing_kind="one_off", activate=False)
winner_runs_between(monkeypatch, tables.CharacteristicRepository, "rows",
lambda: must(api("POST", "/products/SCREW-1/classify-product", CATALOGUE, {"class_code": "SCREWS", "values": [
{"characteristic_code": "MATERIAL", "value": "A2"}]})))
status, body = api("POST", "/products/SCREW-1/activate-product", CATALOGUE, {})
assert status == 409, (status, body)
# ── MUST-2: a dated change reaching an active product today ───────────────────────────────────
def test_must2_a_template_role_from_today_never_reaches_an_active_variant_today():
"""PRD-29: a role change is a new row from a later date, never for today; only a DRAFT product's first rows may start today.
set_unit_role checks the status of the SKU it was called through, not of the variants a template row reaches."""
must(api("POST", "/templates/COLA/generate-variants", CATALOGUE, {"product_type_code": "LEGACY",
"name": "Cola", "base_unit_code": "LTR", "tax_category": "standard", "billing_kind": "one_off", "invoicing_policy": "ordered",
"attributes": [{"name": "volume", "values": [{"value": "0.33 L", "code": "033"}, {"value": "0.75 L", "code": "075"}]}],
"units_of": [{"unit_code": "PK24", "numerator": 18, "denominator": 1, "packaging_level": "case", "einvoice_unit_code": "XPK"}]}))
for pid in ("COLA-033", "COLA-075"):
must(api("POST", f"/products/{pid}/set-units", CATALOGUE, {"sales_unit_code": "LTR"}))
must(api("POST", "/products/COLA-075/activate-product", CATALOGUE, {}))
before = must(api("GET", "/products/COLA-075", CATALOGUE))["roles"]["warehouse"]
status, _ = api("POST", "/products/COLA-033/set-unit-role", CATALOGUE, {"role": "warehouse", "unit_code": "PK24",
"valid_from": "2026-11-20", "scope": "template"})
after = must(api("GET", "/products/COLA-075", CATALOGUE))["roles"]["warehouse"]
assert status == 409 or after == before, (status, before, after)
# ── MUST-3: exact figures, bounded numbers ───────────────────────────────────────────────────
def bolt():
mk_product("BOLT", name="Bolt", base_unit_code="C62", billing_kind="one_off", activate=False)
must(api("POST", "/products/BOLT/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [
{"unit_code": "PK24", "contains_unit": "C62", "contains_quantity": "24", "einvoice_unit_code": "XPK"}]}))
must(api("POST", "/products/BOLT/activate-product", CATALOGUE, {}))
def test_must3_a_surcharge_per_bottle_on_a_pack_price_is_24_surcharges():
"""1.00 per C62 and a formula rule +0.10 per C62: a PK24 (24 C62) costs 24 x 1.10 = 26.40. pricing._apply multiplies the
surcharge by stated / here (1/24) instead of here / stated (24)."""
bolt()
mk_list("LIST-GBP", default_from=DEFAULT, rows=[row("BOLT", "2026-10-02", "1.00", unit_code="C62")],
rules=[{"seq": 1, "applies_to": "product", "target": "BOLT", "method": "formula", "discount_pct": "0",
"surcharge_amount": "0.10", "unit_code": "C62"}])
per_bottle = must(price("BOLT", unit_code="C62", quantity="1"))
assert per_bottle["amount"] == "1.10", per_bottle
pack = must(price("BOLT", unit_code="PK24", quantity="1"))
assert pack["amount"] == "26.40", (pack["unit_price"], pack["pricing_unit_code"], pack["amount"])
def test_guard_a_base_unit_price_is_re_expressed_in_the_asked_unit_only_when_exact_at_four_decimals():
"""PRD-31 rule (mutation-checked: '10000' -> '100', and dropping the check, both fail this test): 1.0004 per LTR is 0.7503 per
C62 (exact): re-expressed; 1.0001 per LTR is 0.750075 per C62 (not exact at 4): the row's own unit and price are kept."""
for pid, p in (("OIL-A", "1.0004"), ("OIL-B", "1.0001")):
mk_product(pid, name="Oil", base_unit_code="LTR", billing_kind="one_off", activate=False)
must(api("POST", f"/products/{pid}/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [
{"unit_code": "C62", "numerator": 3, "denominator": 4}]}))
must(api("POST", f"/products/{pid}/activate-product", CATALOGUE, {}))
mk_list("LIST-GBP", default_from=DEFAULT, rows=[row("OIL-A", "2026-10-02", "1.0004", unit_code="LTR"), row("OIL-B", "2026-10-02", "1.0001", unit_code="LTR")])
a = must(price("OIL-A", unit_code="C62", quantity="4"))
assert (a["unit_price"], a["pricing_unit_code"], a["pricing_quantity"], a["amount"]) == ("0.7503", "C62", 1, "3.00"), a
b = must(price("OIL-B", unit_code="C62", quantity="4"))
assert (b["unit_price"], b["pricing_unit_code"], b["pricing_quantity"], b["amount"]) == ("1.0001", "LTR", 1, "3.00"), b
@pytest.mark.parametrize("product, uom", [({"weight": "0,79"}, {}), ({}, {"relative_factor": "abc"}), ({}, {"rounding": "abc"}),
({}, {"relative_factor": "1e999999"})], ids=["weight", "factor", "rounding", "huge-factor"])
def test_must3_odoo_numeric_text_that_is_not_a_number_stops_the_run_with_a_422(product, uom):
"""PRD-33: a run that cannot be migrated stops with the unit or product listed and nothing written. Decimal('0,79') and
fraction_within('abc') raise decimal.InvalidOperation, which nothing turns into a refusal."""
run = {"db": "odoo19", "cutover": "2026-11-01", "pack_codes": {"Pack of 24": "PK24"},
"uoms": [{"odoo_id": 1, "name": "Units"}, {"odoo_id": 2, "name": "Pack of 24", "relative_uom_id": 1, "relative_factor": "24.0",
"rounding": "1", **uom}],
"products": [{"odoo_id": 100, "default_code": "COLA-075", "type": "consu", "name": "Cola", "uom_id": 1, "uoms": [{"uom_id": 1}, {"uom_id": 2}],
**product}]}
try:
status, body = api("POST", "/migrations/odoo/units", IMPORTER, run)
except Exception as e: # noqa: BLE001 — the in-process transport re-raises a 500
status, body = 500, repr(e)
assert status == 422, (status, body)
def test_must3_a_site_quantity_beyond_numeric_19_4_is_refused():
"""reorder_point is bounded below 1e12 in the asked unit, then multiplied by the unit's factor: 999 999 999 999 PK24 of 18 LTR
is 1.8e13 LTR (fits) but a pallet of 1 440 LTR is 1.44e15, beyond NUMERIC(19,4)'s 15 integer digits."""
cola(activate=False)
must(api("POST", "/products/COLA-075/set-units", CATALOGUE, {"units_of": [
{"unit_code": "PALLET80", "contains_unit": "PK24", "contains_quantity": "80", "einvoice_unit_code": "XPX"}]}))
must(api("POST", "/products/COLA-075/activate-product", CATALOGUE, {}))
status, body = api("POST", "/products/COLA-075/sites/WH1", PLANNER, {"valid_from": "2026-11-21", "planning_method": "reorder_point",
"reorder_point": "999999999999", "unit_code": "PALLET80"})
assert status == 422, (status, body)
# ── MUST-6: a caller's signed response shape ─────────────────────────────────────────────────
def test_must6_mark_unit_used_refusal_carries_the_gate_callers_signed_for():
mk_product("BOX", name="Box", base_unit_code="C62", billing_kind="one_off", activate=False)
must(api("POST", "/products/BOX/set-units", CATALOGUE, {"sales_unit_code": "C62", "units_of": [
{"unit_code": "PK6", "contains_unit": "C62", "contains_quantity": "6"}]}))
must(api("POST", "/products/BOX/activate-product", CATALOGUE, {}))
status, body = api("POST", "/products/BOX/units/PK6/used", SELLING, {"document_ref": "PO-1"})
assert status == 409 and body.get("gate") == "einvoice_unit_code" and "reason" in body and "fix" in body, body
# ── MUST-8: statement count on a request path grows with the catalogue ───────────────────────
def _sited(n, start=0):
for i in range(start, start + n):
pid = f"P-{i:03d}"
mk_product(pid, name=pid, base_unit_code="C62", billing_kind="one_off")
must(api("POST", f"/products/{pid}/sites/WH1", PLANNER, {"valid_from": "2026-11-21", "planning_method": "reorder_point",
"reorder_point": "10"}))
def test_must8_product_sites_costs_the_same_statements_for_5_or_50_products(engine):
_sited(5)
with statements(engine) as s5:
assert len(must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-21", "planning": "active"}))) == 5
_sited(45, start=5)
with statements(engine) as s50:
assert len(must(api("GET", "/product-sites", SALES, {"site_id": "WH1", "on": "2026-11-21", "planning": "active"}))) == 50
assert len(s50) <= len(s5) + 3, (len(s5), len(s50))
def test_guard_characteristic_search_costs_the_same_statements_for_5_or_50_classified_products(engine):
screws()
def screw(i, material):
pid = f"S-{i:03d}"
mk_product(pid, name=pid, base_unit_code="C62", billing_kind="one_off", activate=False)
must(api("POST", f"/products/{pid}/classify-product", CATALOGUE, {"class_code": "SCREWS", "values": [
{"characteristic_code": "MATERIAL", "value": material}, {"characteristic_code": "THREAD", "value": "M4"},
{"characteristic_code": "LENGTH", "value": "20"}]}))
for i in range(5):
screw(i, "A2")
q = {"char": ["MATERIAL:=A2", "LENGTH:>=2 CMT"], "limit": 5}
with statements(engine) as s5:
assert len(must(api("GET", "/products", CATALOGUE, q))["products"]) == 5
for i in range(5, 50):
screw(i, "A4") # unrelated: not A2
with statements(engine) as s50:
assert len(must(api("GET", "/products", CATALOGUE, q))["products"]) == 5
assert len(s50) <= len(s5) + 3, (len(s5), len(s50))
def _odoo(n):
return {"db": "odoo19", "cutover": "2026-11-01",
"uoms": [{"odoo_id": 1, "name": "Units"}, {"odoo_id": 2, "name": "Pack of 24", "relative_uom_id": 1, "relative_factor": "24.0", "rounding": "1"}],
"pack_codes": {"Pack of 24": "PK24"},
"products": [{"odoo_id": 100 + i, "default_code": f"OD-{i:03d}", "type": "product", "name": f"Odoo {i}", "uom_id": 1, "weight": "0.5",
"uoms": [{"uom_id": 1}, {"odoo_id": 1000 + i, "uom_id": 2}]} for i in range(n)]}
def test_must8_the_odoo_migration_and_its_map_do_not_grow_a_statement_per_product(engine):
with statements(engine) as m5:
must(api("POST", "/migrations/odoo/units", IMPORTER, _odoo(5)))
with statements(engine) as again5:
must(api("POST", "/migrations/odoo/units", IMPORTER, _odoo(5)))
run50 = _odoo(50)
run50["db"] = "odoo19b"
with statements(engine) as m50:
must(api("POST", "/migrations/odoo/units", IMPORTER, run50))
with statements(engine) as map5:
must(api("GET", "/migration-map/odoo19b/product.product", SELLING, {"ids": ",".join(str(100 + i) for i in range(5))}))
with statements(engine) as map50:
must(api("GET", "/migration-map/odoo19b/product.product", SELLING, {"ids": ",".join(str(100 + i) for i in range(50))}))
counts = {"migrate5": len(m5), "rerun5": len(again5), "migrate50": len(m50), "map5": len(map5), "map50": len(map50)}
assert len(m50) <= 2 * len(m5) and len(map50) <= len(map5) + 3, counts
def test_a_commit_time_unique_conflict_is_a_409_never_a_500():
"""On DataK3 the loser of a claim race (GTIN, code series) can fail at COMMIT, after the use case returned: the brick's own
repository maps that 23505 to the retry-safe 409 and lets any other integrity failure through (found by the live check)."""
from contextlib import contextmanager
from erp_platform.bricks import Refused
from sqlalchemy.exc import IntegrityError
from brick_product.application import use_cases
class Engine:
def __init__(self, text):
self.text = text
@contextmanager
def begin(self):
yield object()
raise IntegrityError("COMMIT", {}, Exception(self.text))
with pytest.raises(Refused) as lost:
with use_cases._on_engine(Engine("duplicate key value violates unique constraint (SQLSTATE 23505)")):
pass
assert lost.value.status == 409
with pytest.raises(IntegrityError):
with use_cases._on_engine(Engine("null value in column violates not-null constraint")):
pass