dodil/erp-platform

tests/test_compose.py

"""The composition check refuses bricks that would collide — tested with small throwaway bricks."""
import sys
import textwrap

import pytest

from erp_platform import compose


def make_brick(root, package, brick_name, tools, tables, version="0.1.0", seams=(), provides=(), requires=(),
               capability_seams=(), claims_tables=True):
    pkg = root / package
    pkg.mkdir()
    (pkg / "__init__.py").write_text("")
    from erp_platform import capabilities
    (pkg / "schema.sql").write_text("\n".join(
        f"CREATE TABLE IF NOT EXISTS {t} (id INTEGER NOT NULL, PRIMARY KEY (id));" for t in tables)
        + ("\n" + capabilities.ddl(brick_name) if provides and claims_tables else ""))
    body = [f"from erp_platform.bricks import Brick", f"from erp_platform.seam import Seam",
            f"from erp_platform.capabilities import CapabilitySeam",
            f"B = Brick({brick_name!r}, version={version!r}, requires={tuple(requires)!r})"]
    body += [f"C{i} = CapabilitySeam({c!r})" for i, c in enumerate(capability_seams)]
    for i, c in enumerate(provides):
        body += [f"@B.capability({c!r})", f"def cap{i}(call):", f"    return None, {{}}, ''"]
    body += [f"S{i} = Seam({s!r}, {s.upper() + '_BASE_URL'!r}, roles=[], permissions=[])" for i, s in enumerate(seams)]
    for t in tools:
        body += [f"@B.use_case(kind='query', route='GET /{t}')",
                 f"def {t}(caller):", f"    '''{t}'''", "    return {}"]
    (pkg / "use_cases.py").write_text("\n".join(body) + "\n")


@pytest.fixture
def bricks(tmp_path, monkeypatch):
    monkeypatch.syspath_prepend(str(tmp_path))
    yield tmp_path
    for name in list(sys.modules):
        if name.startswith("zz_"):
            del sys.modules[name]


def composition(tmp, *members, tier=1, capabilities=None):
    lines = ["name: t", "version: 0.1.0", f"tier: {tier}", "bricks:"]
    lines += [f"  - {{ name: {n}, brick: '{pkg}.use_cases:B', version: {v} }}" for n, pkg, v in members]
    if capabilities:
        lines += ["capabilities:"] + [f"  {k}: {v}" for k, v in capabilities.items()]
    path = tmp / "compose.yaml"
    path.write_text("\n".join(lines))
    return path


def test_a_clean_composition_serves_one_mcp_with_every_tool(bricks):
    make_brick(bricks, "zz_a", "a", ["list_a"], ["a_rows"], seams=["b"])
    make_brick(bricks, "zz_b", "b", ["list_b"], ["b_rows"], seams=["missing"])
    r = compose.check(compose.load(composition(bricks, ("a", "zz_a", "0.1.0"), ("b", "zz_b", "0.1.0"))))
    assert r["problems"] == [] and r["tools"] == 2
    assert {(s["from"], s["to"], s["present"]) for s in r["seams"]} == {("a", "b", True), ("b", "missing", False)}


def test_two_bricks_claiming_one_table_are_refused(bricks):
    make_brick(bricks, "zz_a", "a", ["list_a"], ["shared_edges"])
    make_brick(bricks, "zz_b", "b", ["list_b"], ["shared_edges"])
    r = compose.check(compose.load(composition(bricks, ("a", "zz_a", "0.1.0"), ("b", "zz_b", "0.1.0"))))
    assert r["problems"] == ["table 'shared_edges' is claimed by both a and b"]


def test_two_bricks_offering_one_tool_name_are_refused(bricks):
    make_brick(bricks, "zz_a", "a", ["get_record"], ["a_rows"])
    make_brick(bricks, "zz_b", "b", ["get_record"], ["b_rows"])
    with pytest.raises(compose.CompositionRefused, match="tool 'get_record' is offered by both a and b"):
        compose.app(composition(bricks, ("a", "zz_a", "0.1.0"), ("b", "zz_b", "0.1.0")))


def test_a_version_that_is_not_installed_is_refused(bricks):
    make_brick(bricks, "zz_a", "a", ["list_a"], ["a_rows"], version="0.2.0")
    r = compose.check(compose.load(composition(bricks, ("a", "zz_a", "0.1.0"))))
    assert r["problems"] == ["a: compose.yaml says 0.1.0, the installed brick is 0.2.0"]


def test_a_brick_that_is_not_installed_says_so(bricks):
    with pytest.raises(compose.CompositionRefused, match="is the brick installed"):
        compose.load(composition(bricks, ("a", "zz_nowhere", "0.1.0")))


def test_tier_two_is_not_served_as_one_app(bricks):
    make_brick(bricks, "zz_a", "a", ["list_a"], ["a_rows"])
    with pytest.raises(compose.CompositionRefused, match="erp-compose plan"):
        compose.app(composition(bricks, ("a", "zz_a", "0.1.0"), tier=2))


def test_routes_are_prefixed_so_two_bricks_paths_never_meet(bricks):
    make_brick(bricks, "zz_a", "a", ["items"], ["a_rows"])
    make_brick(bricks, "zz_b", "b", ["items_b"], ["b_rows"])
    app = compose.app(composition(bricks, ("a", "zz_a", "0.1.0"), ("b", "zz_b", "0.1.0")))
    paths = set(app.openapi()["paths"])
    assert paths == {"/a/items", "/b/items_b", "/composition/process/{brick}"}   # and /items alone is not served
    assert {getattr(r, "path", None) for r in app.routes} >= {"/mcp", "/composition"}


# ── capabilities ────────────────────────────────────────────────────────────────────────────

def wiring(report) -> dict:
    return {c["capability"]: c["provider"] for c in report["capabilities"]}


def test_a_required_capability_with_no_provider_is_allowed_and_reported_manual(bricks):
    make_brick(bricks, "zz_hr", "hr", ["list_runs"], ["hr_runs"], requires=["identity.provision_user"],
               capability_seams=["mail.send"])
    r = compose.check(compose.load(composition(bricks, ("hr", "zz_hr", "0.1.0"))))
    assert r["problems"] == []
    assert wiring(r) == {"identity.provision_user": "manual", "mail.send": "manual"}


def test_one_provider_is_wired_and_registered_for_the_consumers_seam(bricks):
    from erp_platform import capabilities
    make_brick(bricks, "zz_hr", "hr", ["list_runs"], ["hr_runs"], requires=["identity.provision_user"])
    make_brick(bricks, "zz_gws", "gws", ["list_connections"], ["gws_connections"],
               provides=["identity.provision_user", "groups.add_member"])
    path = composition(bricks, ("hr", "zz_hr", "0.1.0"), ("gws", "zz_gws", "0.1.0"))
    r = compose.check(compose.load(path))
    assert r["problems"] == [] and wiring(r) == {"identity.provision_user": "gws", "groups.add_member": "gws"}
    try:
        app = compose.app(path)
        assert capabilities.CapabilitySeam("identity.provision_user").provider() == "gws"
        assert "/gws/capabilities/identity.provision_user" in app.openapi()["paths"]
    finally:
        capabilities.reset_providers()


def test_an_unknown_capability_name_is_refused(bricks):
    make_brick(bricks, "zz_hr", "hr", ["list_runs"], ["hr_runs"], requires=["identity.provision_robot"])
    r = compose.check(compose.load(composition(bricks, ("hr", "zz_hr", "0.1.0"),
                                               capabilities={"calendar.teleport": "manual"})))
    assert r["problems"] == [
        "hr require 'identity.provision_robot', which is not a capability in capabilities.yaml 0.6",
        "compose.yaml maps 'calendar.teleport', which is not a capability in capabilities.yaml 0.6"]


def test_two_providers_need_the_map_to_choose_and_the_other_is_hidden(bricks):
    make_brick(bricks, "zz_gws", "gws", ["list_gws"], ["gws_rows"], provides=["identity.provision_user"])
    make_brick(bricks, "zz_m365", "m365", ["list_m365"], ["m365_rows"], provides=["identity.provision_user"])
    members = (("gws", "zz_gws", "0.1.0"), ("m365", "zz_m365", "0.1.0"))
    r = compose.check(compose.load(composition(bricks, *members)))
    assert r["problems"] == ["identity.provision_user is provided by both gws and m365; choose one under "
                             "capabilities: in compose.yaml"]
    path = composition(bricks, *members, capabilities={"identity.provision_user": "m365"})
    r = compose.check(compose.load(path))
    assert r["problems"] == [] and wiring(r) == {"identity.provision_user": "m365"}
    from erp_platform import capabilities
    try:
        app = compose.app(path)
        paths = set(app.openapi()["paths"])
        assert "/m365/capabilities/identity.provision_user" in paths
        assert "/gws/capabilities/identity.provision_user" not in paths
    finally:
        capabilities.reset_providers()


def test_a_map_entry_must_name_a_brick_that_provides_it(bricks):
    make_brick(bricks, "zz_gws", "gws", ["list_gws"], ["gws_rows"], provides=["groups.add_member"])
    make_brick(bricks, "zz_hr", "hr", ["list_runs"], ["hr_runs"])
    path = composition(bricks, ("gws", "zz_gws", "0.1.0"), ("hr", "zz_hr", "0.1.0"),
                       capabilities={"identity.provision_user": "hr", "groups.add_member": "slack"})
    assert compose.check(compose.load(path))["problems"] == [
        "compose.yaml maps groups.add_member to 'slack', which is not a brick in this composition",
        "compose.yaml maps identity.provision_user to hr, which does not provide it"]


def test_a_provider_without_its_claims_tables_is_refused(bricks):
    make_brick(bricks, "zz_gws", "gws", ["list_gws"], ["gws_rows"], provides=["groups.add_member"],
               claims_tables=False)
    r = compose.check(compose.load(composition(bricks, ("gws", "zz_gws", "0.1.0"))))
    assert len(r["problems"]) == 1 and "gws_capability_claims, gws_capability_log" in r["problems"][0]


def test_plan_prints_the_capability_wiring(bricks, capsys):
    make_brick(bricks, "zz_hr", "hr", ["list_runs"], ["hr_runs"],
               requires=["identity.provision_user", "mail.send"])
    make_brick(bricks, "zz_gws", "gws", ["list_gws"], ["gws_rows"], provides=["identity.provision_user"])
    path = composition(bricks, ("hr", "zz_hr", "0.1.0"), ("gws", "zz_gws", "0.1.0"))
    assert compose.main(["plan", str(path), "--unsigned", "wiring only: no customer design here"]) == 0
    out = capsys.readouterr().out
    assert "capability identity.provision_user → gws  (required by hr)" in out
    assert "capability mail.send → manual" in out
    assert "CAPABILITY_IDENTITY_PROVISION_USER=gws" in out and "GWS_BASE_URL" in out
    assert "CAPABILITY_MAIL_SEND=manual" in out


def test_seams_and_capability_seams_are_found_anywhere_in_the_brick_package(bricks):
    """brick-lead declares its seams in application/neighbours.py: they must be seen without a re-import."""
    make_brick(bricks, "zz_lead", "lead", ["list_leads"], ["lead_rows"])
    app_pkg = bricks / "zz_lead" / "application"
    app_pkg.mkdir()
    (app_pkg / "__init__.py").write_text("")
    (app_pkg / "neighbours.py").write_text(
        "from erp_platform.seam import Seam\nfrom erp_platform.capabilities import CapabilitySeam\n"
        "PARTNERS = Seam('bp', 'BP_BASE_URL', roles=[], permissions=[])\n"
        "DEALS = Seam('pipeline', 'PIPELINE_BASE_URL', roles=[], permissions=[])\n"
        "MAIL = CapabilitySeam('mail.send')\n")
    (app_pkg / "needs_a_bucket.py").write_text("import os\nBUCKET = os.environ['NO_SUCH_BUCKET_VAR']\n")
    (bricks / "zz_lead" / "app.py").write_text("raise RuntimeError('an app module is never imported by check')\n")
    make_brick(bricks, "zz_bp", "bp", ["list_bp"], ["bp_rows"])
    comp = compose.load(composition(bricks, ("lead", "zz_lead", "0.1.0"), ("bp", "zz_bp", "0.1.0")))
    r = compose.check(comp)
    assert r["problems"] == []
    assert {(s["from"], s["to"], s["present"]) for s in r["seams"]} == {("lead", "bp", True), ("lead", "pipeline", False)}
    assert "mail.send" in comp.member("lead").requires()


def test_a_seam_re_imported_into_the_brick_module_counts_once(bricks):
    make_brick(bricks, "zz_lead", "lead", ["list_leads"], ["lead_rows"])
    (bricks / "zz_lead" / "neighbours.py").write_text(
        "from erp_platform.seam import Seam\nPARTNERS = Seam('bp', 'BP_BASE_URL', roles=[], permissions=[])\n")
    with open(bricks / "zz_lead" / "use_cases.py", "a") as f:
        f.write("from zz_lead.neighbours import PARTNERS  # noqa: F401 — brick-lead's workaround\n")
    r = compose.check(compose.load(composition(bricks, ("lead", "zz_lead", "0.1.0"))))
    assert [(s["from"], s["to"]) for s in r["seams"]] == [("lead", "bp")]