dodil/erp-brick-timesheets

DODIL ERP brick: timesheets

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (timesheets v1, signed 2026-10-25 by seemo, erp-method signoff/timesheets):
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_timesheets" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "timesheets" / "module.yaml"
SIGNED = METHOD_DIR / "modules" / "timesheets" / "signoff" / "timesheets"
# the signed files' sha256 (signoff/timesheets/<process>.json, file_sha256), pinned so a changed signature is noticed here
PINNED = {"timesheet": "8cc7d537ed6f198466994b46525c1c036a463e2803fb52432ed58481195ad183"}
PROCESSES = ("timesheet", "billing_run", "ts_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"timesheets_{entity}.bpmn").read_text()
        meta = json.loads((SIGNED / f"timesheets_{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"timesheets_{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": "timesheet"}))
    text = json.dumps(got)
    for key in ("open_timesheet", "submit_timesheet", "approve_project_lines", "approve_timesheet", "return_timesheet", "reopen_timesheet"):
        assert f"Task_{key}" in text, key


def test_every_transition_of_the_slice_is_a_tool_of_the_same_name():
    """The tool a person's agent calls is named for the signed step, so the process and the doors cannot drift apart."""
    from brick_timesheets.application.use_cases import TS
    names = {u.name for u in TS.use_cases}
    for entity in ("timesheet", "billing_run", "ts_migration_run"):
        flow = next(w for w in served()["workflows"] if w["entity"] == entity)
        assert {t["key"] for t in flow["transitions"]} <= names, entity


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"] = "send_week"           # 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 == "timesheet"]
    assert "Task_submit_timesheet" not in bpmn.to_bpmn(one)