dodil/erp-brick-payroll

DODIL ERP brick: payroll

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (payroll v1, signed 2026-10-25 by seemo, erp-method
signoff/payroll): 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."""
import hashlib
import json
import os
import pathlib

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

HERE = pathlib.Path(__file__).resolve().parents[1]
PROCESS = HERE / "src" / "brick_payroll" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC = METHOD_DIR / "modules" / "payroll" / "module.yaml"
SIGNED = METHOD_DIR / "modules" / "payroll" / "signoff" / "payroll"
# the signed files' sha256 (signoff/payroll/<process>.json, file_sha256), pinned so a changed signature is noticed here
PINNED = {"pay_run": "310194c23abc376c50f75a7eec2ea30b68755a0152ccb5d9117848a54a03118e",
          "payroll_employee": "9ee0fd37dbda27c2dc71ea46e24cd95ca4696592d4dd5ebc2355b159acbbb1c6",
          "pension_membership": "acb45ae6c7a48ca3cb3acf196465e964f79526b5b7e5c3f20038e2cd007dce31",
          "prl_ytd_run": "59a08cf237dad218f3384782ad081a637873c2833eb66ecfd5c62e29b8d56e0d"}


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, pinned in PINNED.items():
        signed = (SIGNED / f"payroll_{entity}.bpmn").read_text()
        meta = json.loads((SIGNED / f"payroll_{entity}.json").read_text())
        xml = drawn(entity)
        assert meta["file_sha256"] == pinned, entity
        assert hashlib.sha256(xml.encode()).hexdigest() == pinned, 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():
    got = must(api("GET", "/process/bpmn", ADMIN, {"process_id": "pay_run"}))
    for key in ("open_pay_run", "calculate_pay_run", "approve_approver_payslip", "approve_pay_run", "cancel_pay_run"):
        assert f"Task_{key}" in got["bpmn"], key


def test_a_changed_process_is_not_the_signed_one():
    changed = served()
    changed["workflows"][0]["transitions"][4]["key"] = "sign_off_pay_run"          # 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 == "pay_run"]
    assert "Task_approve_pay_run" not in bpmn.to_bpmn(one)