"""erp-compose bpmn: a module's workflows as BPMN 2.0, from a module spec or an ERP_<MODULE>.md — byte-stable,
XSD-valid, and with ids from keys so a new step adds only its own elements."""
import pathlib
import pytest
import xmlschema
import yaml
from erp_platform import bpmn, compose, signoff
DATA = pathlib.Path(__file__).parent / "data"
XSD = xmlschema.XMLSchema(str(pathlib.Path(__file__).parent / "fixtures" / "bpmn20" / "BPMN20.xsd"))
MD = DATA / "ERP_INVOICING.md"
QUOTE = DATA / "quote.module.yaml"
def valid(xml: str) -> None:
errors = [str(e).splitlines()[0] for e in XSD.iter_errors(xml)]
assert not errors, errors[:3]
@pytest.mark.parametrize("src", [QUOTE, MD], ids=["module.yaml", "ERP_X.md"])
def test_both_inputs_give_xsd_valid_bpmn_with_diagram_interchange(src):
xml = bpmn.generate(src)
valid(xml)
assert "<bpmndi:BPMNShape" in xml and "<bpmndi:BPMNEdge" in xml
ids = signoff.elements(xml)
assert all(i.replace("Task_", "Task_") in ids for i in ids)
def test_the_same_input_gives_the_same_bytes(tmp_path):
a, b = bpmn.generate(MD), bpmn.generate(MD)
assert a == b
assert compose.main(["bpmn", "--from", str(MD), "--out", str(tmp_path / "p.bpmn")]) == 0
assert (tmp_path / "p.bpmn").read_text() == a
def test_ids_follow_the_stable_scheme():
ids = set(signoff.elements(bpmn.generate(MD)))
assert {"Task_approve_invoice", "Gateway_approve_invoice__if", "Gateway_invoice_draft__if",
"Timer_send_reminder", "Process_invoicing_invoice", "Flow_Task_send_reminder__End_dunning_reminded"} - ids \
<= {"Flow_Task_send_reminder__End_dunning_reminded"}
xml = bpmn.generate(MD)
assert 'dodil:step="approve_invoice"' in xml and 'dodil:approval="true"' in xml
assert 'id="Process_invoicing_invoice_v1"' in xml and "<bpmn:serviceTask" in xml
def _with_step(spec: dict, step: dict) -> dict:
spec = yaml.safe_load(yaml.safe_dump(spec))
spec["workflows"][0]["states"].append(step["to"])
spec["workflows"][0]["transitions"].append(step)
return spec
def test_adding_a_step_only_adds_or_changes_its_own_elements():
base = yaml.safe_load(QUOTE.read_text())
before = bpmn.to_bpmn(bpmn.from_module(base, str(QUOTE)))
# sent already had one exit: the new exit adds the state's choice gateway and re-routes into it
after = bpmn.to_bpmn(bpmn.from_module(_with_step(base, {"key": "expire_quote", "from": "sent", "to": "expired",
"by": "system", "timer": "P30D"}), str(QUOTE)))
rows = signoff.diff(before, after)
added = {r["id"] for r in rows if r["op"] == "added"}
assert {"Task_expire_quote", "Timer_expire_quote", "Gateway_quote_sent__if", "End_quote_expired"} <= added
touched = {r["id"] for r in rows if r["op"] != "added"}
# what else moved is only the flow into `sent`'s old single exit, and the lane set / process text
assert all(i.startswith("Flow_") or i.startswith("Lane_") or i.startswith("Process_") for i in touched), touched
assert not [r for r in rows if r["op"] == "removed" and not r["id"].startswith("Flow_")]
# every task that existed is unchanged
assert not [r for r in rows if r["kind"] == "task" and r["op"] != "added"]
def test_a_step_at_a_state_that_already_chooses_adds_only():
base = yaml.safe_load(QUOTE.read_text())
two = _with_step(base, {"key": "expire_quote", "from": "sent", "to": "expired", "by": "rep"})
three = _with_step(two, {"key": "withdraw_quote", "from": "sent", "to": "withdrawn", "by": "rep"})
rows = signoff.diff(bpmn.to_bpmn(bpmn.from_module(two, "q")), bpmn.to_bpmn(bpmn.from_module(three, "q")))
assert {r["op"] for r in rows if r["kind"] not in ("process",)} == {"added"}, rows
def test_every_module_spec_in_the_method_draws(tmp_path):
method = pathlib.Path(__file__).parents[2] / "erp-method" / "modules"
specs = [p for p in method.glob("*/module.yaml") if "transitions" in p.read_text()]
if not specs:
pytest.skip("erp-method is not checked out beside erp-platform")
for p in specs:
valid(bpmn.generate(p))
def test_one_workflow_and_an_unknown_one():
xml = bpmn.generate(MD, "dunning")
assert "Task_escalate" in xml and "Task_approve_invoice" not in xml
with pytest.raises(bpmn.WorkflowError, match="no workflow 'nope'"):
bpmn.generate(MD, "nope")
@pytest.mark.parametrize("edit, line, says", [
(("| escalate | reminded | escalated |", "| escalate | reminded | gone |"), 31, "gone not in States"),
(("| escalate | reminded | escalated | biller |", "| escalate | reminded | escalated | auditor |"), 31,
"auditor is not an actor"),
(("| cancel_draft | draft | cancelled | biller | | no number used | | |", "| cancel_draft | draft | cancelled |"),
22, "3 cells, the header has 8"),
(("| Key | From | To | Who | Gate / requires | Effect | Timer | Approval |",
"| Key | From | To | Owner | Gate / requires | Effect | Timer | Approval |"), 17, "unknown column 'Owner'"),
(("| void_invoice |", "| draft_invoice |"), 23, "key 'draft_invoice' is used twice"),
(("States: open, reminded, escalated\nTerminal: escalated\n", ""), 28, "write `States:"),
(("Terminal: escalated\n", ""), 30, "state 'escalated' has no exit and is not terminal"),
(("Terminal: escalated", "Terminal: closed"), 27, "Terminal closed not in States"),
])
def test_a_malformed_table_fails_with_its_line(tmp_path, edit, line, says):
bad = tmp_path / "ERP_INVOICING.md"
bad.write_text(MD.read_text().replace(*edit, 1))
with pytest.raises(bpmn.WorkflowError) as e:
bpmn.generate(bad)
assert f"ERP_INVOICING.md:{line}:" in str(e.value) and says in str(e.value)
def test_the_cli_names_the_line_and_exits_1(tmp_path, capsys):
bad = tmp_path / "ERP_X.md"
bad.write_text("# x\n\n## Workflows\n### quote\n| From | To |\n|---|---|\n| none | draft |\n")
assert compose.main(["bpmn", "--from", str(bad)]) == 1
assert "ERP_X.md:5: the table needs columns From, To and Who" in capsys.readouterr().err
PROJECTS = DATA / "ERP_PROJECTS.md"
def test_a_step_that_keeps_its_state_is_sequenced_by_row_order_and_invents_no_state():
xml = bpmn.generate(PROJECTS)
valid(xml)
ids = set(signoff.elements(xml))
# draft: open → send_brief → (fill_usage, To empty = From, optional by its gate) → start_project, in row order
assert {"Flow_Task_open_project__Task_send_brief", "Flow_Task_send_brief__Gateway_fill_usage__if",
"Flow_Gateway_fill_usage__if_end__Task_start_project"} <= ids
assert not [i for i in ids if i.startswith(("End_project_draft", "Gateway_project_draft", "Gateway_project_delivered"))]
# no self-loop, and every row is a task
assert not [i for i in ids if i.startswith("Flow_") and i.split("__")[0][5:] == i.split("__")[1]]
assert {f"Task_{k}" for k in ("send_brief", "fill_usage", "review_delivery")} <= ids
def test_an_external_actor_waits_in_its_own_lane_with_a_message_and_is_signed():
xml = bpmn.generate(PROJECTS)
el = signoff.elements(xml)
assert el["Task_send_brief"]["tag"] == "receiveTask" and el["Task_send_brief"]["fields"]["dodil:external"] == "true"
assert el["Task_send_brief"]["fields"]["lane"] == "Lane_project_client_contact"
assert el["Participant_client_contact"]["kind"] == "participant"
assert el["Flow_Participant_client_contact__Task_send_brief"]["tag"] == "messageFlow"
# part of what is signed: dropping the external marker changes the hash, and the diff names the step
plain = PROJECTS.read_text().replace("Client contact (external)", "Client contact")
other = bpmn.parse_md(plain, str(PROJECTS))
changed = bpmn.to_bpmn(other)
assert signoff.content_hash(changed) != signoff.content_hash(xml)
rows = {r["id"]: r for r in signoff.diff(xml, changed)}
assert rows["Task_send_brief"]["op"] == "changed" and rows["Participant_client_contact"]["op"] == "removed"
def test_module_yaml_marks_external_actors_too():
spec = yaml.safe_load(QUOTE.read_text())
spec["actors"][1]["external"] = True
xml = bpmn.to_bpmn(bpmn.from_module(spec, str(QUOTE)))
valid(xml)
assert signoff.elements(xml)["Task_accept_quote"]["tag"] == "receiveTask"
def test_a_terminal_state_with_exits_may_end_or_go_on():
el = signoff.elements(bpmn.generate(MD))
# issued is terminal AND has an exit (void): a choice between the end and voiding
assert el["Flow_Gateway_invoice_issued__if__End_invoice_issued"]["fields"]["name"] == "done"
assert "Flow_Gateway_invoice_issued__if__Task_void_invoice" in el
assert "Flow_Task_approve_invoice__Gateway_invoice_issued__if" not in el # reached via the outcome gateway
assert "Flow_Gateway_approve_invoice__if__Gateway_invoice_issued__if" in el
assert "End_invoice_void" in el and "Gateway_invoice_void__if" not in el # terminal, no exits: just an end
def test_a_self_step_with_a_gate_is_optional_and_rejoins():
el = signoff.elements(bpmn.generate(PROJECTS))
assert el["Flow_Gateway_fill_usage__if__Task_fill_usage"]["fields"]["name"] == "the draft has a usage line"
assert el["Flow_Gateway_fill_usage__if__Gateway_fill_usage__if_end"]["fields"]["name"] == "otherwise"
assert {"Flow_Task_send_brief__Gateway_fill_usage__if", "Flow_Task_fill_usage__Gateway_fill_usage__if_end",
"Flow_Gateway_fill_usage__if_end__Task_start_project"} <= set(el)
assert "Gateway_send_brief__if" not in el # no gate: unconditional