dodil/erp-brick-payments

DODIL ERP brick: payments

tests/test_process.py

"""The process this brick serves is the process the spec's workflows say (payments version 4, signed 2026-10-23, erp-method
v0.1.27). This test holds application/process.yaml equal to the spec's `workflows:` (and its actors); a process change goes
through `erp-compose change propose` and a new signature first. A missing erp-method checkout is an error, never a skip."""
import os
import pathlib

import yaml
from erp_platform import bpmn

HERE = pathlib.Path(__file__).resolve().parents[1]
PROCESS = HERE / "src" / "brick_payments" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "payments" / "module.yaml"
KEEP = ("key", "name", "from", "to", "by", "gate", "effect", "requires", "approval", "timer")


def test_the_served_process_is_the_specs_workflows():
    assert SPEC.exists(), f"the signed spec is needed: set ERP_METHOD_DIR (tried {SPEC})"
    spec, served = yaml.safe_load(SPEC.read_text()), yaml.safe_load(PROCESS.read_text())
    assert served["workflows"] == [
        {"entity": w["entity"], "states": w["states"], "terminal": w["terminal"],
         "transitions": [{k: t[k] for k in KEEP if k in t} for t in w["transitions"]]} for w in spec["workflows"]]
    assert [a["id"] for a in served["actors"]] == [a["id"] for a in spec["actors"]]


def test_every_entity_draws_as_bpmn_with_a_task_per_transition():
    data = yaml.safe_load(PROCESS.read_text())
    for w in data["workflows"]:
        one = bpmn.from_module(data, "module.yaml")
        one.workflows = [x for x in one.workflows if x.entity == w["entity"]]
        xml = bpmn.to_bpmn(one)
        for t in w["transitions"]:
            assert f"Task_{t['key']}" in xml, (w["entity"], t["key"])


def test_a_changed_process_is_not_the_specs():
    data = yaml.safe_load(PROCESS.read_text())
    data["workflows"][0]["transitions"][0]["name"] = "Add whatever we like"
    assert data["workflows"] != yaml.safe_load(SPEC.read_text())["workflows"]