dodil/erp-brick-expenses

DODIL ERP brick: expenses

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (expenses v1, signed 2026-10-25 by seemo, erp-method signoff/expenses):
process.yaml is module.yaml's workflows, copied, and GET /process/bpmn draws it. The step ids are frozen: the drawn BPMN of each process
is byte-for-byte the signed file (its sha256 is pinned here too). 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 hashlib
import json
import os
import pathlib

import yaml
from conftest import EMPLOYEE, api, must
from erp_platform import bpmn

HERE = pathlib.Path(__file__).resolve().parents[1]
PROCESS = HERE / "src" / "brick_expenses" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "expenses" / "module.yaml"
SIGNED = METHOD_DIR / "modules" / "expenses" / "signoff" / "expenses"
# the signed files' sha256 (signoff/expenses/<process>.json, file_sha256), pinned so a changed signature is noticed here
PINNED = {"expense_report": "fedadeb614898029c7f6c3b711ae821e30446470bea15c1027de1c5c2efdede7"}
PROCESSES = ("expense_report", "card_txn", "exp_migration_run")


def served():
    return yaml.safe_load(PROCESS.read_text())


def drawn(entity):
    one = bpmn.from_module(served(), "module.yaml")
    one.workflows = [w for w in one.workflows if w.entity == entity]
    return bpmn.to_bpmn(one)


def test_the_served_process_is_the_signed_one():
    assert SPEC.exists(), f"the signed spec is needed: set ERP_METHOD_DIR (tried {SPEC})"
    spec = yaml.safe_load(SPEC.read_text())
    assert (served()["id"], served()["version"]) == (spec["id"], spec["version"])
    assert served()["workflows"] == spec["workflows"]                     # every state, step id, gate and effect, as signed
    assert [a["id"] for a in served()["actors"]] == [a["id"] for a in spec["actors"]]


def test_every_process_is_drawn_as_the_signed_file_under_its_frozen_ids():
    for entity in PROCESSES:
        signed = (SIGNED / f"expenses_{entity}.bpmn").read_text()
        meta = json.loads((SIGNED / f"expenses_{entity}.json").read_text())
        xml = drawn(entity)
        assert hashlib.sha256(xml.encode()).hexdigest() == meta["file_sha256"], entity
        assert xml == signed, entity                                       # every Task_, Flow_ and Gateway_ id, as the customer signed it
        for t in next(w for w in served()["workflows"] if w["entity"] == entity)["transitions"]:
            assert f"Task_{t['key']}" in xml, (entity, t["key"])
    for entity, pinned in PINNED.items():
        assert json.loads((SIGNED / f"expenses_{entity}.json").read_text())["file_sha256"] == pinned


def test_the_brick_serves_its_process_at_get_process_bpmn():
    got = must(api("GET", "/process/bpmn", EMPLOYEE, {"process_id": "expense_report"}))
    text = json.dumps(got)
    for key in ("create_report", "submit_report", "approve_level", "complete_approval", "post_report", "confirm_paid"):
        assert f"Task_{key}" in text, key


def test_a_changed_process_is_not_the_signed_one():
    changed = served()
    changed["workflows"][0]["transitions"][0]["name"] = "Add whatever we like"
    assert changed["workflows"] != yaml.safe_load(SPEC.read_text())["workflows"]
    changed = served()
    changed["workflows"][0]["transitions"][2]["key"] = "send_report"          # a renamed step is a new id: the signed one is gone
    one = bpmn.from_module(changed, "module.yaml")
    one.workflows = [w for w in one.workflows if w.entity == "expense_report"]
    assert "Task_submit_report" not in bpmn.to_bpmn(one)