dodil/erp-platform

tests/test_permissions.py

"""0.9.2 permissions: `module:entity:action` with segment-wise wildcards, machine permissions on an exact grant only,
no role-name fallback, and what `erp-compose check` says about a composition's roles."""
import json
import logging
import sys

import pytest

from erp_platform import auth, compose, permissions_check as pc
from erp_platform.bricks import Brick
from erp_platform.scenarios.personas import RoleCatalogue


def who(*grants, roles=()):
    return {"sub": "x", "email": "x@co.example", "permissions": list(grants), "roles": list(roles)}


@pytest.mark.parametrize("grant", ["pj:item:read", "pj:item:*", "pj:*:read", "pj:*:*", "*:*:*", "*:item:read",
                                   "*:*:read"])
def test_a_grant_covers_a_permission_segment_by_segment(grant):
    assert auth.has_permission(who(grant), "pj:item:read")


@pytest.mark.parametrize("grant", ["pj:item:write", "pj:project:*", "so:*:*", "*:*:write", "pj:*:write"])
def test_a_grant_that_differs_in_any_segment_does_not(grant):
    assert not auth.has_permission(who(grant), "pj:item:read")


@pytest.mark.parametrize("grant", ["*", "*:*", "*:*:*:*", "pj:*", "pj:it*:read", "PJ:item:read", "pj:item:read ",
                                   " *:*:*", "", "pj:item1:read", "pj:it-em:read", "pj::read", 7, None])
def test_a_malformed_grant_passes_nothing_and_never_raises(grant):
    assert not auth.has_permission(who(grant), "pj:item:read")


def test_a_malformed_grant_is_logged_once(caplog):
    with caplog.at_level(logging.WARNING, logger="erp_platform.auth"):
        for _ in range(3):
            auth.has_permission(who("zz:q*:once"), "zz:qq:once")
    assert sum("zz:q*:once" in r.message for r in caplog.records) == 1


def test_permissions_that_are_not_a_list_grant_nothing():
    # a string would answer `in` by substring: "pj:item:read_all" contains "pj:item:read"
    assert not auth.has_permission({"permissions": "pj:item:read_all", "roles": []}, "pj:item:read")
    assert not auth.has_permission({"permissions": {"pj:item:read": 1}, "roles": []}, "pj:item:read")


def test_a_required_permission_that_is_not_concrete_passes_no_one():
    assert not auth.has_permission(who("*:*:*", "TODO(permission):pj:x:y"), "TODO(permission):pj:x:y")


def test_the_role_name_fallback_is_gone():
    assert not auth.has_permission(who(roles=["pj:item:read"]), "pj:item:read")


def test_a_use_case_or_route_requiring_a_wildcard_is_refused_at_registration():
    b = Brick("zz_wild")
    with pytest.raises(ValueError, match="concrete"):
        b.use_case(kind="query", route="GET /x", permission="zz_wild:item:*")
    with pytest.raises(ValueError, match="concrete"):
        auth.require_permission("*:*:*")


def test_a_machine_permission_passes_on_an_exact_grant_only():
    assert auth.ACT_FOR in auth.machine_permissions()
    for g in ("*:*:*", "erp:*:*", "erp:seam:*", "*:seam:act_for", "erp:*:act_for"):
        assert not auth.has_permission(who(g), auth.ACT_FOR), g
    assert auth.has_permission(who(auth.ACT_FOR), auth.ACT_FOR)


def test_a_brick_declares_its_machine_permissions():
    Brick("zz_mach", machine_permissions=("zz_mach:source:link",))
    assert auth.is_machine("zz_mach:source:link")
    assert not auth.has_permission(who("*:*:*", "zz_mach:*:*"), "zz_mach:source:link")
    assert auth.has_permission(who("zz_mach:source:link"), "zz_mach:source:link")
    assert auth.has_permission(who("zz_mach:*:*"), "zz_mach:source:read")      # its person permissions still match
    with pytest.raises(ValueError, match="concrete"):
        Brick("zz_bad", machine_permissions=("zz_bad:*:link",))


def test_the_scenario_role_catalogue_uses_the_same_rule():
    cat = RoleCatalogue.load({"roles": {"boss": ["*:*:*"], "ops": ["pj:*:read"], "pj:item:read": []}})
    assert cat.holds("boss", "pj:item:write") and not cat.holds("boss", auth.ACT_FOR)
    assert cat.holds("ops", "pj:item:read") and not cat.holds("ops", "pj:item:write")
    assert not cat.holds("pj:item:read", "pj:item:read")                      # a role's NAME grants nothing



def test_seeing_every_row_is_a_permission_when_the_brick_says_which():
    reader = {"email": "a@co.example", "roles": ["manager"], "permissions": ["pj:item:read"]}
    assert auth.owner_scope(reader, see_all="pj:item:read_all") == "a@co.example"   # the role name is not read
    boss = {"email": "b@co.example", "roles": [], "permissions": ["pj:*:read_all"]}
    assert auth.owner_scope(boss, see_all="pj:item:read_all") is None
    assert auth.owner_scope(boss, mine=True, see_all="pj:item:read_all") == "b@co.example"


def test_effective_expands_a_role_against_the_catalog():
    cat = ["pj:item:read", "pj:item:accept", "so:order:read", auth.ACT_FOR]
    assert auth.effective(["pj:*:*"], cat) == ["pj:item:accept", "pj:item:read"]
    assert auth.effective(["*:*:*"], cat) == ["pj:item:accept", "pj:item:read", "so:order:read"]
    assert auth.effective(["*:*:*", auth.ACT_FOR], cat)[0] == auth.ACT_FOR


# ── erp-compose check ─────────────────────────────────────────────────────────────────────────

BRICK = '''from erp_platform import auth
from erp_platform.bricks import Brick
B = Brick("zz_shop", machine_permissions=("zz_shop:tick:run",))


@B.use_case(kind="query", route="GET /orders", permission="zz_shop:order:read")
def list_orders(caller):
    """list"""
    return {"scope": auth.owner_scope(caller)}


@B.use_case(kind="command", route="POST /orders/{order_id}/approve", permission="zz_shop:order:approve")
def approve_order(caller, order_id: str):
    """approve"""
    if "manager" in caller["roles"]:
        return {}
    ok = caller.get("roles")  # role-check: the pre-0.9.2 demo pool, removed with it
    return {}


@B.use_case(kind="command", route="POST /orders/{order_id}", permission="zz_shop:order:write", destructive=True)
def delete_order(caller, order_id: str):
    """delete"""
    partner = {"roles": ["customer"]}
    return {"customer": "customer" in partner.get("roles", [])}


@B.use_case(kind="command", route="POST /tick", permission="zz_shop:tick:run", tool=False)
def tick(caller):
    """tick"""
    return {}


@B.use_case(kind="command", route="POST /link", permission="zz_shop:order:link", tool=False)
def link(caller):
    """seam-only, not declared machine"""
    return {}
'''


@pytest.fixture
def shop(tmp_path, monkeypatch):
    pkg = tmp_path / "zz_shop"
    pkg.mkdir()
    (pkg / "__init__.py").write_text("")
    (pkg / "schema.sql").write_text("CREATE TABLE IF NOT EXISTS zz_shop_orders (id TEXT NOT NULL, PRIMARY KEY (id));")
    (pkg / "use_cases.py").write_text(BRICK)
    monkeypatch.syspath_prepend(str(tmp_path))
    (tmp_path / "compose.yaml").write_text(
        "name: t\nversion: 0.1.0\ntier: 1\nbricks:\n  - { name: zz_shop, brick: 'zz_shop.use_cases:B', version: 0.1.0 }\n"
        "  # the roles below are the scenario's\n")
    yield tmp_path
    for name in list(sys.modules):
        if name.startswith("zz_shop"):
            del sys.modules[name]


def roles_file(tmp, roles, services=()):
    p = tmp / "roles.yaml"
    p.write_text(json.dumps({"roles": roles, "service_clients": list(services)}))
    return p


def test_check_prints_each_roles_effective_permissions_and_warns_on_a_wildcard_reaching_approval(shop, capsys):
    roles_file(shop, {"clerk": ["zz_shop:order:*"], "approver": ["zz_shop:order:approve", "zz_shop:order:read"],
                      "admin": ["*:*:*"], "ticker": ["zz_shop:tick:run"], "odd": ["zz_shop:or*:read"]},
               services=["ticker"])
    assert compose.main(["check", str(shop / "compose.yaml")]) == 0
    out, err = capsys.readouterr()
    assert "clerk          zz_shop    zz_shop:order:approve, zz_shop:order:link, zz_shop:order:read, zz_shop:order:write" in out
    assert "admin          zz_shop    zz_shop:order:approve, zz_shop:order:link, zz_shop:order:read, zz_shop:order:write" in out
    assert "ticker         zz_shop    zz_shop:tick:run" in out and "odd: nothing" in out
    assert "role clerk: wildcard zz_shop:order:* reaches the approval command zz_shop.approve_order" in err
    assert "role clerk: wildcard zz_shop:order:* reaches the destructive command zz_shop.delete_order" in err
    assert "role admin: wildcard *:*:* reaches the approval command" in err
    assert "role approver" not in err                                    # named explicitly: no warning
    assert "role odd: grant 'zz_shop:or*:read' is malformed" in err
    assert "zz_shop.link: a seam-only command behind zz_shop:order:link, which is not a machine permission" in err
    assert "zz_shop.tick" not in err


def test_a_machine_permission_in_a_person_role_is_a_warning(shop, capsys):
    roles_file(shop, {"clerk": ["zz_shop:tick:run"]})
    compose.main(["check", str(shop / "compose.yaml")])
    assert "role clerk: holds the machine permission(s) zz_shop:tick:run" in capsys.readouterr().err


def test_role_name_decisions_in_a_brick_are_warned_and_refused_under_strict(shop, capsys):
    roles_file(shop, {"clerk": []})
    assert compose.main(["check", str(shop / "compose.yaml")]) == 0
    err = capsys.readouterr().err
    flagged = [line for line in err.splitlines() if "decides by permission, not a role name" in line]
    assert len(flagged) == 2, flagged                       # owner_scope() without see_all, caller["roles"]
    assert any("use_cases.py:9:" in f for f in flagged) and any("use_cases.py:15:" in f for f in flagged)
    assert compose.main(["check", "--strict", str(shop / "compose.yaml")]) == 1
    assert "PROBLEM a brick decides by permission" in capsys.readouterr().err


def test_with_no_role_catalogue_check_says_so(shop, capsys):
    assert compose.main(["check", str(shop / "compose.yaml")]) == 0
    assert "roles: no role catalogue" in capsys.readouterr().out


def test_an_appid_role_catalog_json_is_read_as_is(shop, capsys):
    p = shop / "pool.json"
    p.write_text(json.dumps({"sales": ["zz_shop:order:read"]}))
    assert compose.main(["check", "--roles", str(p), str(shop / "compose.yaml")]) == 0
    assert "sales          zz_shop    zz_shop:order:read" in capsys.readouterr().out