dodil/erp-brick-pos

DODIL ERP brick: point of sale

tests/test_process.py

"""The process this brick serves is the process the customer SIGNED (pos v1, signed 2026-10-25, erp-method v0.1.29 signoff/pos).
application/process.yaml is module.yaml's workflows, copied, and this test holds the two equal; the drawing of each entity is held
to the sha256 the customer signed. 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 os
import pathlib

import yaml
from conftest import api
from erp_platform import bpmn, signoff

HERE = pathlib.Path(__file__).resolve().parents[1]
PROCESS = HERE / "src" / "brick_pos" / "application" / "process.yaml"
METHOD_DIR = pathlib.Path(os.environ.get("ERP_METHOD_DIR") or HERE.parent / "erp-method")
SPEC_DIR = METHOD_DIR / "modules" / "pos"
SPEC = yaml.safe_load((SPEC_DIR / "module.yaml").read_text())
SERVED = yaml.safe_load(PROCESS.read_text())

# The steps slice A implements as use cases, by their frozen ids: open_session, record_cash_move, count_session, approve_difference
# (the session) and ring_up, take_payment, finalise_receipt (the receipt). The others are in test_remaining.py.
BUILT = {"open_session", "record_cash_move", "count_session", "approve_difference", "ring_up", "take_payment", "finalise_receipt"}


def test_the_served_process_is_the_signed_one():
    assert SPEC_DIR.exists(), f"the signed spec is needed: set ERP_METHOD_DIR (tried {SPEC_DIR})"
    assert (SERVED["id"], SERVED["name"], SERVED["version"]) == (SPEC["id"], SPEC["name"], SPEC["version"]) == ("pos", SPEC["name"], 1)
    assert SERVED["actors"] == SPEC["actors"]
    assert SERVED["workflows"] == SPEC["workflows"]                   # every state, step id, gate and effect, as signed


def _drawn(entity: str) -> str:
    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_every_entity_is_drawn_exactly_as_the_customer_signed_it():
    for wf in SPEC["workflows"]:
        signed = json.loads((SPEC_DIR / "signoff" / "pos" / f"pos_{wf['entity']}.json").read_text())
        assert signoff.matches(signed["bpmn_sha256"], _drawn(wf["entity"])), f"{wf['entity']} no longer draws as signed"


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


def test_a_changed_process_is_not_the_signed_one():
    """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"]
    changed = yaml.safe_load(PROCESS.read_text())
    changed["workflows"][0]["transitions"][0]["key"] = "open_the_till"
    one = bpmn.from_module(changed, "module.yaml")
    one.workflows = [w for w in one.workflows if w.entity == "session"]
    signed = json.loads((SPEC_DIR / "signoff" / "pos" / "pos_session.json").read_text())
    assert not signoff.matches(signed["bpmn_sha256"], bpmn.to_bpmn(one))


def test_the_process_is_served_as_bpmn(nb):
    from conftest import ACCOUNTANT
    status, body = api("GET", "/process/bpmn", ACCOUNTANT)
    assert status == 200 and "Task_open_session" in json.dumps(body) and "Task_finalise_receipt" in json.dumps(body)


def test_the_steps_built_are_signed_steps():
    keys = {t["key"] for w in SPEC["workflows"] for t in w["transitions"]}
    assert BUILT <= keys