dodil/erp-platform

tests/test_process_view.py

"""The brick contract (0.8): every brick shows its process as BPMN — `declare_process` serves show_process_bpmn,
brick-hr's show_playbook_bpmn counts — and `erp-compose check` refuses a brick built for 0.8 with none, warns for
one pinned older, and prints a `process_view: none  # reason` waiver."""
import anyio
import httpx
import pytest

from erp_platform import compose, seam
from test_compose import bricks, make_brick  # noqa: F401 — the fixture and the helper

WORKFLOW = ("\nB.declare_process({'actors': [{'id': 'rep', 'name': 'Sales rep'}], 'workflows': [{'entity': 'quote', "
            "'states': ['draft', 'sent'], 'transitions': [{'key': 'draft_quote', 'from': 'none', 'to': 'draft', "
            "'by': 'rep'}, {'key': 'send_quote', 'from': 'draft', 'to': 'sent', 'by': 'rep'}]}]})\n")


def with_process(root, package, name):
    make_brick(root, package, name, [f"list_{name}"], [f"{name}_rows"])
    p = root / package / "use_cases.py"
    p.write_text(p.read_text() + WORKFLOW)


def comp(tmp, *rows):
    lines = ["name: t", "version: 0.1.0", "tier: 1", "bricks:"] + [f"  {r}" for r in rows]
    path = tmp / "compose.yaml"
    path.write_text("\n".join(lines) + "\n")
    return path


def test_a_declared_process_is_a_route_and_one_composition_tool(bricks):
    with_process(bricks, "zz_a", "a")
    with_process(bricks, "zz_b", "b")
    path = comp(bricks, "- { name: a, brick: 'zz_a.use_cases:B', version: 0.1.0 }",
                "- { name: b, brick: 'zz_b.use_cases:B', version: 0.1.0 }")
    r = compose.check(compose.load(path))
    assert r["problems"] == [] and r["warnings"] == []
    assert r["process_views"] == [{"brick": "a", "view": "show_process_bpmn"}, {"brick": "b", "view": "show_process_bpmn"}]
    app = compose.app(path)
    me = {"sub": "u1", "email": "u@x.test", "app_roles": ["admin"], "permissions": []}

    async def go():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://local") as c:
            h = {**seam.gateway_headers(me), "Accept": "application/json"}
            return (await c.get("/a/process/bpmn", headers=h), await c.get("/composition/process/b", headers=h),
                    await c.get("/composition/process/a", params={"process_id": "nope"}, headers=h))
    own, via, missing = anyio.run(go)
    assert own.status_code == 200 and "Task_send_quote" in own.json()["bpmn"]
    assert via.status_code == 200 and via.json()["brick"] == "b" and len(via.json()["content_sha256"]) == 64
    assert missing.status_code == 404


def test_no_process_view_warns_for_a_brick_pinned_older_and_refuses_one_built_for_0_8(bricks, monkeypatch):
    make_brick(bricks, "zz_a", "a", ["list_a"], ["a_rows"])
    path = comp(bricks, "- { name: a, brick: 'zz_a.use_cases:B', version: 0.1.0 }")
    monkeypatch.setattr(compose, "platform_floor", lambda m: (0, 5, 2))
    r = compose.check(compose.load(path))
    assert r["problems"] == [] and len(r["warnings"]) == 1
    assert "a: declares no process view" in r["warnings"][0] and "built for erp-platform 0.5.2" in r["warnings"][0]
    monkeypatch.setattr(compose, "platform_floor", lambda m: (0, 8, 0))
    r = compose.check(compose.load(path))
    assert len(r["problems"]) == 1 and "a: declares no process view" in r["problems"][0]   # the refusal


def test_a_waiver_needs_its_reason_and_is_printed(bricks, monkeypatch, capsys):
    monkeypatch.setattr(compose, "platform_floor", lambda m: (0, 8, 0))
    make_brick(bricks, "zz_a", "a", ["list_a"], ["a_rows"])
    path = comp(bricks, "- { name: a, brick: 'zz_a.use_cases:B', version: 0.1.0, process_view: none }   "
                        "# master data only: no lifecycle")
    r = compose.check(compose.load(path))
    assert r["problems"] == [] and r["process_views"] == [{"brick": "a", "view": None,
                                                           "waived": "master data only: no lifecycle"}]
    assert compose.main(["check", str(path)]) == 0
    assert "process a: WAIVED — master data only: no lifecycle" in capsys.readouterr().out
    bare = comp(bricks, "- { name: a, brick: 'zz_a.use_cases:B', version: 0.1.0, process_view: none }")
    assert "needs its reason" in compose.check(compose.load(bare))["problems"][0]


def test_a_malformed_declared_process_fails_at_import():
    from erp_platform.bricks import Brick
    from erp_platform.bpmn import WorkflowError
    with pytest.raises(WorkflowError):
        Brick("x").declare_process({"workflows": [{"entity": "q", "transitions": [{"from": "a", "to": "b"}]}]})