dodil/erp-brick-manufacturing

DODIL ERP brick: manufacturing

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (manufacturing v1, signed 2026-10-25): application/process.yaml is
module.yaml's actors and workflows, copied, step ids frozen - and the BPMN it draws hashes to the signed files. A change to the
process goes through `erp-compose change propose` and a new signature first, then the code. A missing erp-method checkout is an
error, never a skip."""
import json
import pathlib

import yaml
from conftest import METHOD_DIR, OPERATOR, api, must
from erp_platform import bpmn, signoff

PROCESS = pathlib.Path(__file__).resolve().parents[1] / "src" / "brick_manufacturing" / "application" / "process.yaml"
MODULE = pathlib.Path(METHOD_DIR) / "modules" / "manufacturing"
SPEC = yaml.safe_load((MODULE / "module.yaml").read_text())


def _drawn(entity):
    data = yaml.safe_load(PROCESS.read_text())
    one = bpmn.from_module(data, "module.yaml")
    one.workflows = [x for x in one.workflows if x.entity == entity]
    return bpmn.to_bpmn(one)


def test_the_served_process_is_the_signed_one():
    served = yaml.safe_load(PROCESS.read_text())
    assert (served["id"], served["name"], served["version"]) == (SPEC["id"], SPEC["name"], SPEC["version"])
    assert served["actors"] == SPEC["actors"]
    assert served["workflows"] == SPEC["workflows"]                  # every state, step id, gate and effect, as signed


def test_every_signed_step_is_drawn_under_its_signed_id():
    for wf in SPEC["workflows"]:
        drawn = _drawn(wf["entity"])
        for t in wf["transitions"]:
            assert f"Task_{t['key']}" in drawn, (wf["entity"], t["key"])


def test_the_drawn_bpmn_is_the_signed_file():
    for rec_file in sorted((MODULE / "signoff" / "manufacturing").glob("manufacturing_*.json")):
        rec = json.loads(rec_file.read_text())
        assert signoff.matches(rec["bpmn_sha256"], _drawn(rec["source"]["workflow"]), rec["source"]["from"]), rec_file.name


def test_a_changed_process_does_not_match():
    """The guard is real: one renamed step and the comparison fails."""
    changed = yaml.safe_load(PROCESS.read_text())
    changed["workflows"][0]["transitions"][0]["name"] = "Add whatever we like"
    assert changed["workflows"] != SPEC["workflows"]


def test_the_process_is_served_at_get_process_bpmn():
    body = must(api("GET", "/process/bpmn", OPERATOR))
    text = json.dumps(body)
    assert "Task_release_order" in text and "Task_close_order" in text