"""Generated probes (erp-compose scaffold 0.9.2): WHO may run each command, in WHICH state, with WHICH side
effect — from module.yaml workflows (from/to/by/terminal), actors' permissions and seams. Every one passes: no
xfail is left.
The tables are the spec, not the code. Three were completed by hand because the call needs them, never to make a
probe pass: a body (reactivate needs its reason, rename what it changes, assign_customer its customer), the
permission assign_customer is gated by (product:customer_price_list:write, the pricing manager's own), and a
seeded product that satisfies the activation gate. SOD is empty on purpose: a second person is a SETTING
(publish_needs_second_person, off by default), proved in test_rules.py; it is not unconditional.
"""
import json
import pytest
from conftest import api, person
from erp_platform import auth, seam
from brick_product.application import use_cases
from brick_product.domain.price_list import PriceList
from brick_product.domain.product import Product
CLASSES = {"product": Product, "price_list": PriceList}
PLURAL = {'product': 'products', 'price_list': 'price_lists'}
STATE_FIELD = {'product': 'status', 'price_list': 'status'}
# A seeded row must satisfy the gates the workflow puts on its transitions: a product is activated only when it has a
# unit, a tax category, a billing type and an invoicing policy (PRD-02), so the one seeded to be activated has them.
FILL = {'product': {'name': 'Probe product', 'base_unit_code': 'C62', 'sales_unit_code': 'C62', 'tax_category': 'standard', 'billing_kind': 'one_off',
'billing_plan': 'single', 'invoicing_policy': 'ordered'}}
KEYS = {'product': {'product_id': 'PRODUCT-ID-1'}, 'price_list': {'price_list_id': 'PRICE-LIST-ID-1'}}
STATES = {'product': ['draft', 'active', 'archived'], 'price_list': ['draft', 'published', 'retired']}
TERMINAL = {'product': ['archived'], 'price_list': ['retired']}
OWNED = [] # roots with an owner: row visibility by owner
PARENTS = {} # root -> (parent root, its key on the child)
ROLES = {'catalogue_manager': ['product:product:read', 'product:product:write', 'product:product:archive', 'product:category:write', 'product:settings:write'], 'pricing_manager': ['product:product:read', 'product:price_list:read', 'product:price_list:write', 'product:price_list:publish', 'product:customer_price_list:write', 'product:settings:write'], 'owner': ['product:product:read', 'product:price_list:read', 'product:settings:controls'], 'sales_rep': ['product:product:read', 'product:price_list:read'], 'system': ['product:product:read', 'product:price_list:read', 'product:catalogue:import', 'product:product:suggest', 'product:price_list:check']}
READ_ALL = {} # root -> the permission that sees every owner's rows (never a role)
NS = 'product'
# Outbound seams (module.yaml seams) -> the transitions that may fire them. PRD-20 (quote asks find_price) and PRD-21
# (an order or invoice line copies the product) are INBOUND: the other brick calls this one's routes. PRD-22 is an import
# through this brick's own use case. The one thing this brick asks a neighbour is whether a customer exists, and only
# assign_customer does (business-partner; a customer cannot be assigned without it — tests/test_seams.py).
SEAMS = {'business-partner': ['assign_customer']}
# Separation of duties: (the request, the approval) — the person who did the first may not do the second.
SOD = []
COMMANDS = [
{'name': 'ai_check_duplicate', 'root': 'product', 'route': 'POST /products/{product_id}/ai-check-duplicate', 'permission': 'product:product:suggest', 'frm': ['draft'], 'to': ['draft'], 'by': ['system'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'ai_suggest_details', 'root': 'product', 'route': 'POST /products/{product_id}/ai-suggest-details', 'permission': 'product:product:suggest', 'frm': ['draft'], 'to': ['draft'], 'by': ['system'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'set_units', 'root': 'product', 'route': 'POST /products/{product_id}/set-units', 'permission': 'product:product:write', 'frm': ['draft'], 'to': ['draft'], 'by': ['catalogue_manager'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'add_unit', 'root': 'product', 'route': 'POST /products/{product_id}/add-unit', 'permission': 'product:product:write', 'frm': ['active'], 'to': ['active'], 'by': ['catalogue_manager'], 'body': {'unit': {'unit_code': 'XBX', 'numerator': 6, 'denominator': 1}}, 'loop': True, 'idempotent': False},
{'name': 'complete_product', 'root': 'product', 'route': 'POST /products/{product_id}/complete-product', 'permission': 'product:product:write', 'frm': ['draft'], 'to': ['draft'], 'by': ['catalogue_manager'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'activate_product', 'root': 'product', 'route': 'POST /products/{product_id}/activate-product', 'permission': 'product:product:write', 'frm': ['draft'], 'to': ['active'], 'by': ['catalogue_manager'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'rename_product', 'root': 'product', 'route': 'POST /products/{product_id}/rename-product', 'permission': 'product:product:write', 'frm': ['active'], 'to': ['active'], 'by': ['catalogue_manager'], 'body': {'name': 'Renamed'}, 'loop': True, 'idempotent': False},
{'name': 'archive_product', 'root': 'product', 'route': 'POST /products/{product_id}/archive-product', 'permission': 'product:product:archive', 'frm': ['active'], 'to': ['archived'], 'by': ['catalogue_manager'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'reactivate_product', 'root': 'product', 'route': 'POST /products/{product_id}/reactivate-product', 'permission': 'product:product:write', 'frm': ['archived'], 'to': ['active'], 'by': ['catalogue_manager'], 'body': {'reason': 'because'}, 'loop': False, 'idempotent': True},
{'name': 'set_prices', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/set-prices', 'permission': 'product:price_list:write', 'frm': ['draft'], 'to': ['draft'], 'by': ['pricing_manager'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'ai_price_check', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/ai-price-check', 'permission': 'product:price_list:check', 'frm': ['draft'], 'to': ['draft'], 'by': ['system'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'publish_price_list', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/publish-price-list', 'permission': 'product:price_list:publish', 'frm': ['draft'], 'to': ['published'], 'by': ['pricing_manager'], 'body': {}, 'loop': False, 'idempotent': True},
{'name': 'change_prices', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/change-prices', 'permission': 'product:price_list:write', 'frm': ['published'], 'to': ['published'], 'by': ['pricing_manager'], 'body': {}, 'loop': True, 'idempotent': False},
{'name': 'assign_customer', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/assign-customer', 'permission': 'product:customer_price_list:write', 'frm': ['published'], 'to': ['published'], 'by': ['pricing_manager'], 'body': {'bp_id': '7311'}, 'loop': True, 'idempotent': True},
{'name': 'retire_price_list', 'root': 'price_list', 'route': 'POST /price-lists/{price_list_id}/retire-price-list', 'permission': 'product:price_list:write', 'frm': ['published'], 'to': ['retired'], 'by': ['pricing_manager'], 'body': {}, 'loop': False, 'idempotent': True},
]
SUSPECTS = ['owner', 'created_at', 'created_by', 'approved_by', 'approved_at', 'roles', 'permissions', 'is_admin', 'tenant']
BY = {c["name"]: c for c in COMMANDS}
REFUSED = (403, 404, 409)
def who(role, email=None):
# the role's PERMISSIONS and no role name: a brick decides by permission, so the probes carry none
return person(email or f"{role}@shop.example", ROLES.get(role, []))
ALL_PERMS = sorted({p for perms in ROLES.values() for p in perms} | {c["permission"] for c in COMMANDS})
BOSS = person("boss@shop.example", ALL_PERMS)
def seed(root, state, owner="boss@shop.example", parent_state=None):
"""A row of `root` in `state` (and its parent in `parent_state`), written through the repository."""
if root in PARENTS and parent_state:
seed(PARENTS[root][0], parent_state, owner)
agg = CLASSES[root](**KEYS[root], **FILL.get(root, {}))
setattr(agg, STATE_FIELD[root], state)
if root in OWNED:
agg.owner = owner
if root in PARENTS:
parent, fk = PARENTS[root]
setattr(agg, fk, next(iter(KEYS[parent].values())))
with use_cases.repository() as repo:
getattr(repo, PLURAL[root]).save(agg, 0)
def state_of(root):
with use_cases.repository() as repo:
found = getattr(repo, PLURAL[root]).get(*KEYS[root].values())
return getattr(found[0], STATE_FIELD[root])
def run(c, headers, extra=None):
path = c["route"].split(" ", 1)[1]
for k, v in KEYS[c["root"]].items():
path = path.replace("{" + k + "}", str(v))
return api("POST", path, headers, {**c["body"], **(extra or {})})
@pytest.fixture
def seams(monkeypatch):
"""Every seam call any code makes, recorded (and answered ok) — wherever the call is written."""
calls = []
def record(self, path, *a, **kw):
calls.append((self.name, path))
return seam.Reply(True, 200, data={})
monkeypatch.setattr(seam.Seam, "post", record)
monkeypatch.setattr(seam.Seam, "get", record)
return calls
def todo(reason):
return pytest.mark.xfail(strict=True, raises=AssertionError, reason=reason)
def cases(rows):
return [pytest.param(*r[:-1], marks=r[-1], id="-".join(map(str, r[:-1]))) if r[-1] else
pytest.param(*r[:-1], id="-".join(map(str, r[:-1]))) for r in rows]
# ── the permission matrix: every role × every command ─────────────────────────────────────
def _matrix():
rows = []
for c in COMMANDS:
for role, perms in ROLES.items():
holds, named = c["permission"] in perms, role in c["by"]
if not holds:
rows.append((c["name"], role, "refused", None))
elif not named:
rows.append((c["name"], role, "refused", todo(
f"TODO(permission): {role} holds {c['permission']} but the workflow gives {c['name']} to "
f"{', '.join(c['by'])} — a finer permission, or the role check in the use case")))
else:
rows.append((c["name"], role, "served", None))
for role in c["by"]:
if role not in ROLES:
continue
if c["permission"] not in ROLES[role]:
rows.append((c["name"], role, "served", todo(
f"TODO(permission): the workflow gives {c['name']} to {role}, whose permissions lack "
f"{c['permission']} — fix the spec's actors")))
return rows
@pytest.mark.parametrize("name,role,expect", cases(_matrix()))
def test_the_permission_matrix(name, role, expect, seams):
c = BY[name]
me = f"{role}@shop.example"
seed(c["root"], c["frm"][0], owner=me)
status, body = run(c, who(role, me))
if expect == "refused":
assert status in REFUSED, (status, body)
assert state_of(c["root"]) == c["frm"][0] and seams == []
else:
assert status == 200, (status, body)
# ── wildcard grants (erp_platform.auth 0.9.2): the gate itself, held to every command ────────────
MACHINE = auth.machine_permissions() # exact grant only: no wildcard reaches them
MALFORMED = ["*", "*:*", "*:*:*:*", f"{NS}:*", f"{NS.upper()}:*:*", f"{NS}:i*:*", " *:*:*", "", 7]
def gate(c):
return {"detail": f"missing permission: {c['permission']}"}
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_every_permission_by_wildcard_passes_every_person_command_and_no_machine_one(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
status, body = run(c, person("wild@shop.example", ["*:*:*"]))
if c["permission"] in MACHINE:
assert (status, body) == (403, gate(c))
else:
assert status != 401 and body != gate(c), (status, body)
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_a_module_wildcard_is_confined_to_its_module(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
assert run(c, person("wild@shop.example", ["zz_other:*:*"])) == (403, gate(c))
assert state_of(c["root"]) == c["frm"][0] and seams == []
status, body = run(c, person("wild@shop.example", [f"{NS}:*:*"]))
assert c["permission"] in MACHINE or body != gate(c), (status, body)
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_malformed_grants_pass_nothing(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0], owner="wild@shop.example")
assert run(c, person("wild@shop.example", MALFORMED)) == (403, gate(c))
assert state_of(c["root"]) == c["frm"][0] and seams == []
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS if c["root"] in OWNED])
def test_another_person_of_the_same_role_gets_404(name):
c = BY[name]
role = next((r for r in c["by"] if r in ROLES and READ_ALL.get(c["root"]) not in ROLES[r] and c["permission"] in ROLES[r]), None)
if role is None:
pytest.skip("no person role is given this command with its permission")
seed(c["root"], c["frm"][0], owner=f"{role}@shop.example")
status, _ = run(c, who(role, f"another.{role}@shop.example"))
assert status == 404 # never 403: a 403 would confirm the record exists
assert state_of(c["root"]) == c["frm"][0]
@pytest.mark.parametrize("request_step,approval", [tuple(x) for x in SOD] or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no approval in the workflows"))])
@todo("TODO(rule): separation of duties — who did the request may not approve it")
def test_the_requester_may_not_approve(request_step, approval):
first, second = BY[request_step], BY[approval]
seed(first["root"], first["frm"][0])
assert run(first, BOSS)[0] == 200
status, body = run(second, BOSS)
assert status in REFUSED, (status, body)
# ── state guards: every command in every state its workflow does not allow ─────────────────
def _wrong_states():
rows = []
for c in COMMANDS:
ok = set(c["frm"]) | (set() if c["loop"] else set(c["to"]))
rows += [(c["name"], s, None) for s in STATES[c["root"]] if s not in ok]
return rows
@pytest.mark.parametrize("name,state", cases(_wrong_states()))
def test_a_command_in_a_state_its_workflow_does_not_allow_is_refused(name, state, seams):
c = BY[name]
seed(c["root"], state)
status, body = run(c, BOSS)
assert status == 409, (status, body)
assert state_of(c["root"]) == state and seams == []
def _under_terminal_parent():
rows = []
for c in COMMANDS:
if c["root"] in PARENTS:
parent = PARENTS[c["root"]][0]
rows += [(c["name"], s, todo(f"TODO(rule): {c['name']} under a {s} {parent}")) for s in TERMINAL[parent]]
return rows
@pytest.mark.parametrize("name,parent_state", cases(_under_terminal_parent()) or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no parent with a lifecycle"))])
def test_a_command_under_a_terminal_parent_is_refused(name, parent_state, seams):
c = BY[name]
seed(c["root"], c["frm"][0], parent_state=parent_state)
status, body = run(c, BOSS)
assert status in REFUSED, (status, body)
assert state_of(c["root"]) == c["frm"][0] and seams == []
# ── side effects: a seam fires only on the transitions that name it ────────────────────────
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_a_seam_fires_only_on_the_transitions_that_name_it(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
assert run(c, BOSS)[0] == 200
wrong = [(brick, path) for brick, path in seams if name not in SEAMS.get(brick, [])]
assert wrong == [], f"{name} called a seam the workflow does not give it"
@pytest.mark.parametrize("brick,name", [(b, n) for b, steps in SEAMS.items() for n in steps] or
[pytest.param(None, None, marks=pytest.mark.skip(reason="no outbound seam"))])
def test_the_transition_that_names_a_seam_calls_it(brick, name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
assert run(c, BOSS)[0] == 200
assert [b for b, _ in seams] == [brick]
# ── idempotency and mass assignment ────────────────────────────────────────────────────────
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS if c["idempotent"]])
def test_a_replay_answers_the_same_and_moves_nothing(name, seams):
c = BY[name]
seed(c["root"], c["frm"][0])
first = run(c, BOSS)
assert first[0] == 200, first
calls = list(seams)
assert run(c, BOSS) == first
assert seams == calls # no second side effect
@pytest.mark.parametrize("name", [c["name"] for c in COMMANDS])
def test_fields_it_did_not_ask_for_are_never_applied(name):
c = BY[name]
seed(c["root"], c["frm"][0])
sentinel = "mass-assign-sentinel"
extra = {f: sentinel for f in SUSPECTS + [STATE_FIELD[c["root"]]] if f not in c["body"]}
status, body = run(c, BOSS, extra)
assert status in (200, 422), (status, body)
assert sentinel not in json.dumps(body)
assert state_of(c["root"]) != sentinel