dodil/erp-brick-time-off

DODIL ERP brick: time off

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (time-off v1, signed 2026-10-25 by seemo, erp-method signoff/time-off):
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_time_off" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "time-off" / "module.yaml"
SIGNED = METHOD_DIR / "modules" / "time-off" / "signoff" / "time-off"
# the signed files' sha256 (signoff/time-off/<process>.json, file_sha256), pinned so a changed signature is noticed here
PINNED = {"leave_request": "1c02d3f6743917463c262fe943bb2cdc089ca730638234c42e7fe62d9b46b739",
          "sickness": "49dc421558ccd001261920283d4c49f39ceb7a853e4706770536bd272b176ad9",
          "tof_migration_run": "642e8c9c26f724dd368a902f7645b826ac5744c607d2e0004aa73ac19f94aea0"}
PROCESSES = tuple(PINNED)


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"time-off_{entity}.bpmn").read_text()
        meta = json.loads((SIGNED / f"time-off_{entity}.json").read_text())
        xml = drawn(entity)
        assert hashlib.sha256(xml.encode()).hexdigest() == meta["file_sha256"] == PINNED[entity], 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"])


def test_the_brick_serves_its_process_at_get_process_bpmn():
    for entity in PROCESSES:
        got = must(api("GET", "/process/bpmn", EMPLOYEE, {"process_id": entity}))
        text = json.dumps(got)
        for t in next(w for w in served()["workflows"] if w["entity"] == entity)["transitions"]:
            assert f"Task_{t['key']}" in text, (entity, t["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"][1]["key"] = "agree_leave"           # 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 == "leave_request"]
    assert "Task_approve_leave" not in bpmn.to_bpmn(one)


def test_every_leave_request_arrow_has_a_use_case_with_the_signed_step_id_as_its_name():
    """The slice-A use cases are named for the signed transition keys (their tool names), so the process and the doors cannot drift."""
    from brick_time_off.application import use_cases
    keys = [t["key"] for w in served()["workflows"] if w["entity"] == "leave_request" for t in w["transitions"]]
    for key in keys:
        assert hasattr(use_cases, key), key