bvandeusen
f3e919892d
Implements #253's framework: a small core capability registry (steward/core/capabilities.py) where a module/plugin publishes a named, role-gated action and a consumer discovers it via has_capability() and runs it via invoke_capability() — no hard import, graceful degradation, permission propagation (actor role checked against the capability's required_role). Core publishes "ansible.run_playbook" (operator) wrapping ansible.runner. trigger_run (extended to accept a caller-built inventory). First consumer: the host_agent plugin gains "Deploy via Ansible" on its settings page — pick an inventory target/group and it installs/updates the agent via the bundled host_agent/install.yml, minting a fresh token per host and injecting it as an inventory hostvar (turning per-host curl|sh into one run). Exposed role ordering as middleware.role_meets. Unit tests for the registry + role checks. Task #253 (milestone #37). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""Unit tests for the plugin/core capability registry."""
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from steward.auth.middleware import role_meets
|
|
from steward.core.capabilities import (
|
|
CapabilityForbidden,
|
|
CapabilityUnavailable,
|
|
clear_capabilities,
|
|
get_capability,
|
|
has_capability,
|
|
invoke_capability,
|
|
register_capability,
|
|
)
|
|
from steward.models.users import UserRole
|
|
|
|
|
|
def setup_function():
|
|
clear_capabilities()
|
|
|
|
|
|
def test_register_has_get_defaults_to_admin():
|
|
register_capability("x.do", lambda: "ok", label="Do X")
|
|
assert has_capability("x.do")
|
|
assert get_capability("x.do").label == "Do X"
|
|
assert get_capability("x.do").required_role == UserRole.admin
|
|
|
|
|
|
def test_invoke_sync_callable():
|
|
register_capability("x.sum", lambda a, b: a + b, label="Sum", required_role=UserRole.viewer)
|
|
assert asyncio.run(invoke_capability("x.sum", UserRole.viewer, 2, 3)) == 5
|
|
|
|
|
|
def test_invoke_async_callable():
|
|
async def double(v):
|
|
return v * 2
|
|
|
|
register_capability("x.double", double, label="D", required_role=UserRole.viewer)
|
|
assert asyncio.run(invoke_capability("x.double", UserRole.operator, 4)) == 8
|
|
|
|
|
|
def test_invoke_unknown_capability_raises():
|
|
with pytest.raises(CapabilityUnavailable):
|
|
asyncio.run(invoke_capability("nope", UserRole.admin))
|
|
|
|
|
|
def test_invoke_forbidden_when_role_too_low():
|
|
register_capability("x.admin", lambda: 1, label="A", required_role=UserRole.admin)
|
|
with pytest.raises(CapabilityForbidden):
|
|
asyncio.run(invoke_capability("x.admin", UserRole.operator))
|
|
|
|
|
|
def test_role_meets_ordering():
|
|
assert role_meets(UserRole.admin, UserRole.operator) is True
|
|
assert role_meets(UserRole.operator, UserRole.operator) is True
|
|
assert role_meets(UserRole.viewer, UserRole.operator) is False
|