cleveragents-core/features/steps/permission_system_steps.py

"""Step definitions for the permission system feature.

Tests the permission domain model, PermissionService in local and
server modes, the enforce_permission decorator, and role binding
management.
"""

from __future__ import annotations

import os

from behave import given, then, when
from behave.runner import Context
from pydantic import ValidationError

from cleveragents.application.services.permission_service import (
    PermissionService,
    enforce_permission,
    set_default_permission_service,
)
from cleveragents.domain.models.core.permission import (
    DEFAULT_LOCAL_ROLE_MAPPING,
    ROLE_PERMISSIONS,
    PermissionAction,
    PermissionCheck,
    PermissionPolicy,
    PermissionRole,
    PermissionScope,
    RoleBinding,
)

# ── Helpers ───────────────────────────────────────────────────────

_SERVER_MODE_ENV = "CLEVERAGENTS_SERVER_MODE"


def _set_local_mode() -> None:
    os.environ.pop(_SERVER_MODE_ENV, None)


def _set_server_mode() -> None:
    os.environ[_SERVER_MODE_ENV] = "true"


# ── Enum steps ────────────────────────────────────────────────────


@given("the PermissionRole enum")
def step_given_permission_role_enum(context: Context) -> None:
    context.enum_values = [e.value for e in PermissionRole]


@given("the PermissionScope enum")
def step_given_permission_scope_enum(context: Context) -> None:
    context.enum_values = [e.value for e in PermissionScope]


@given("the PermissionAction enum")
def step_given_permission_action_enum(context: Context) -> None:
    context.enum_values = [e.value for e in PermissionAction]


@then('the enum should contain "{value}"')
def step_enum_should_contain(context: Context, value: str) -> None:
    assert value in context.enum_values, f"Expected '{value}' in {context.enum_values}"


# ── ROLE_PERMISSIONS ──────────────────────────────────────────────


@given("the ROLE_PERMISSIONS constant")
def step_given_role_permissions(context: Context) -> None:
    context.role_permissions = ROLE_PERMISSIONS


@then('the "{role}" role should have {count:d} allowed actions')
def step_role_has_n_actions(context: Context, role: str, count: int) -> None:
    perm_role = PermissionRole(role)
    actions = context.role_permissions[perm_role]
    assert len(actions) == count, (
        f"Expected {count} actions for {role}, got {len(actions)}"
    )


# ── RoleBinding model ────────────────────────────────────────────


@given(
    'a role binding for principal "{principal}" with role "{role}" '
    'on "{scope}" scope "{scope_id}"'
)
def step_given_role_binding(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.binding = binding
    # Also register on service if present
    svc: PermissionService | None = getattr(context, "permission_service", None)
    if svc is not None:
        svc.add_binding(binding)


@then('the binding principal should be "{expected}"')
def step_binding_principal(context: Context, expected: str) -> None:
    assert context.binding.principal == expected


@then('the binding role should be "{expected}"')
def step_binding_role(context: Context, expected: str) -> None:
    assert context.binding.role == expected


@then('the binding scope should be "{expected}"')
def step_binding_scope(context: Context, expected: str) -> None:
    assert context.binding.scope == expected


@then('the binding scope_id should be "{expected}"')
def step_binding_scope_id(context: Context, expected: str) -> None:
    assert context.binding.scope_id == expected


# ── PermissionPolicy ─────────────────────────────────────────────


@given("a default PermissionPolicy")
def step_given_default_policy(context: Context) -> None:
    context.policy = PermissionPolicy()


@given(
    'a PermissionPolicy with an override for "{principal}" as "{role}" '
    'on "{scope}" scope "{scope_id}"'
)
def step_given_policy_with_override(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    override = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.policy = PermissionPolicy(allow_overrides=[override])


@then("default_deny should be true")
def step_default_deny_true(context: Context) -> None:
    assert context.policy.default_deny is True


@then("allow_overrides should be empty")
def step_overrides_empty(context: Context) -> None:
    assert len(context.policy.allow_overrides) == 0


@then("the policy should have {count:d} allow override")
def step_policy_override_count(context: Context, count: int) -> None:
    assert len(context.policy.allow_overrides) == count


# ── PermissionService — local mode ────────────────────────────────


@given("a PermissionService in local mode")
def step_given_service_local(context: Context) -> None:
    _set_local_mode()
    context.permission_service = PermissionService()


@given("a PermissionService in server mode with no bindings")
def step_given_service_server_no_bindings(context: Context) -> None:
    _set_server_mode()
    context.permission_service = PermissionService()


@given("a PermissionService in server mode")
def step_given_service_server(context: Context) -> None:
    _set_server_mode()
    context.permission_service = PermissionService()


@given("a PermissionService in server mode with default_deny disabled")
def step_given_service_server_no_deny(context: Context) -> None:
    _set_server_mode()
    policy = PermissionPolicy(default_deny=False)
    context.permission_service = PermissionService(policy=policy)


@given(
    'a policy override for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"'
)
def step_given_policy_override_on_service(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    override = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    policy = PermissionPolicy(allow_overrides=[override])
    context.permission_service = PermissionService(policy=policy)


# ── Permission check ─────────────────────────────────────────────


@when(
    'I check permission for "{principal}" to "{action}" on "{scope}" scope "{scope_id}"'
)
def step_check_permission(
    context: Context, principal: str, action: str, scope: str, scope_id: str
) -> None:
    svc: PermissionService = context.permission_service
    context.permission_result = svc.check_permission(
        principal=principal,
        action=PermissionAction(action),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )


@then("the permission check result should be true")
def step_check_result_true(context: Context) -> None:
    assert context.permission_result.result is True, (
        f"Expected True, got False: {context.permission_result.reason}"
    )


@then("the permission check result should be false")
def step_check_result_false(context: Context) -> None:
    assert context.permission_result.result is False, (
        f"Expected False, got True: {context.permission_result.reason}"
    )


@then('the reason should contain "{fragment}"')
def step_reason_contains(context: Context, fragment: str) -> None:
    assert fragment in context.permission_result.reason, (
        f"Expected '{fragment}' in '{context.permission_result.reason}'"
    )


# ── DEFAULT_LOCAL_ROLE_MAPPING ────────────────────────────────────


@given("the DEFAULT_LOCAL_ROLE_MAPPING constant")
def step_given_default_local_mapping(context: Context) -> None:
    context.local_mapping = DEFAULT_LOCAL_ROLE_MAPPING


@then('the wildcard entry should map to "{expected}"')
def step_wildcard_maps_to(context: Context, expected: str) -> None:
    assert context.local_mapping["*"] == expected


# ── PermissionCheck model ─────────────────────────────────────────


@given(
    'a PermissionCheck for "{principal}" action "{action}" scope "{scope}" '
    'scope_id "{scope_id}" result true reason "{reason}"'
)
def step_given_permission_check_model(
    context: Context,
    principal: str,
    action: str,
    scope: str,
    scope_id: str,
    reason: str,
) -> None:
    context.check_model = PermissionCheck(
        principal=principal,
        action=PermissionAction(action),
        scope=PermissionScope(scope),
        scope_id=scope_id,
        result=True,
        reason=reason,
    )


@then('the check principal should be "{expected}"')
def step_check_principal(context: Context, expected: str) -> None:
    assert context.check_model.principal == expected


@then('the check action should be "{expected}"')
def step_check_action(context: Context, expected: str) -> None:
    assert context.check_model.action == expected


@then('the check scope should be "{expected}"')
def step_check_scope(context: Context, expected: str) -> None:
    assert context.check_model.scope == expected


@then('the check scope_id should be "{expected}"')
def step_check_scope_id(context: Context, expected: str) -> None:
    assert context.check_model.scope_id == expected


@then("the check result should be true")
def step_check_model_result_true(context: Context) -> None:
    assert context.check_model.result is True


@then('the check reason should be "{expected}"')
def step_check_reason(context: Context, expected: str) -> None:
    assert context.check_model.reason == expected


# ── enforce_permission decorator ──────────────────────────────────


@given(
    "a function decorated with enforce_permission for "
    '"{action}" on "{scope}" scope "{scope_id}"'
)
def step_given_decorated_fn(
    context: Context, action: str, scope: str, scope_id: str
) -> None:
    context.enforce_action = action
    context.enforce_scope = scope
    context.enforce_scope_id = scope_id


@when("I call the decorated function in local mode")
def step_call_decorated_local(context: Context) -> None:
    _set_local_mode()

    @enforce_permission(
        PermissionAction(context.enforce_action),
        PermissionScope(context.enforce_scope),
        context.enforce_scope_id,
    )
    def _guarded() -> str:
        return "ok"

    try:
        context.decorated_result = _guarded()
        context.decorated_error = None
    except PermissionError as exc:
        context.decorated_result = None
        context.decorated_error = exc


@when("I call the decorated function in server mode with no bindings")
def step_call_decorated_server(context: Context) -> None:
    _set_server_mode()

    @enforce_permission(
        PermissionAction(context.enforce_action),
        PermissionScope(context.enforce_scope),
        context.enforce_scope_id,
    )
    def _guarded() -> str:
        return "ok"

    try:
        context.decorated_result = _guarded()
        context.decorated_error = None
    except PermissionError as exc:
        context.decorated_result = None
        context.decorated_error = exc


@then("it should succeed without raising")
def step_no_error(context: Context) -> None:
    assert context.decorated_error is None
    assert context.decorated_result == "ok"


@then("it should raise a PermissionError")
def step_permission_error_raised(context: Context) -> None:
    assert context.decorated_error is not None
    assert isinstance(context.decorated_error, PermissionError)


# ── Binding management ────────────────────────────────────────────


@given('I add a binding for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"')
def step_add_binding(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.permission_service.add_binding(binding)


@when(
    'I remove the binding for "{principal}" as "{role}" on "{scope}" scope "{scope_id}"'
)
def step_remove_binding(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.remove_result = context.permission_service.remove_binding(binding)


@when(
    'I try to remove a binding for "{principal}" as "{role}" '
    'on "{scope}" scope "{scope_id}"'
)
def step_try_remove_binding(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.remove_result = context.permission_service.remove_binding(binding)


@then("the remove result should be false")
def step_remove_result_false(context: Context) -> None:
    assert context.remove_result is False


# ── get_role_bindings ─────────────────────────────────────────────


@when('I get role bindings for "{principal}" on "{scope}" scope "{scope_id}"')
def step_get_role_bindings(
    context: Context, principal: str, scope: str, scope_id: str
) -> None:
    svc: PermissionService = context.permission_service
    context.fetched_bindings = svc.get_role_bindings(
        principal=principal,
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )


@then("I should receive {count:d} binding")
def step_binding_count_singular(context: Context, count: int) -> None:
    assert len(context.fetched_bindings) == count, (
        f"Expected {count} bindings, got {len(context.fetched_bindings)}"
    )


@then("I should receive {count:d} bindings")
def step_binding_count_plural(context: Context, count: int) -> None:
    assert len(context.fetched_bindings) == count, (
        f"Expected {count} bindings, got {len(context.fetched_bindings)}"
    )


@then('the first binding role should be "{expected}"')
def step_first_binding_role(context: Context, expected: str) -> None:
    assert context.fetched_bindings[0].role == expected


# ── Duplicate binding prevention ──────────────────────────────────


@when(
    'I try to add a duplicate binding for "{principal}" as "{role}" '
    'on "{scope}" scope "{scope_id}"'
)
def step_try_add_duplicate_binding(
    context: Context, principal: str, role: str, scope: str, scope_id: str
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.add_result = context.permission_service.add_binding(binding)


@then("the add result should be false")
def step_add_result_false(context: Context) -> None:
    assert context.add_result is False


# ── RoleBinding validation ────────────────────────────────────────


@when("I try to create a role binding with empty principal")
def step_try_create_binding_empty_principal(context: Context) -> None:
    context.permission_error = None
    try:
        RoleBinding(
            principal="",
            role=PermissionRole.VIEWER,
            scope=PermissionScope.PROJECT,
            scope_id="proj-1",
        )
    except ValidationError as exc:
        context.permission_error = exc


@when("I try to create a role binding with empty scope_id")
def step_try_create_binding_empty_scope_id(context: Context) -> None:
    context.permission_error = None
    try:
        RoleBinding(
            principal="alice",
            role=PermissionRole.VIEWER,
            scope=PermissionScope.PROJECT,
            scope_id="",
        )
    except ValidationError as exc:
        context.permission_error = exc


@then("a permission validation error should be raised")
def step_permission_validation_error(context: Context) -> None:
    assert context.permission_error is not None, "Expected a ValidationError"
    assert isinstance(context.permission_error, ValidationError)


# ── PermissionCheck with result=False ─────────────────────────────


@given(
    'a PermissionCheck for "{principal}" action "{action}" scope "{scope}" '
    'scope_id "{scope_id}" result false reason "{reason}"'
)
def step_given_permission_check_model_false(
    context: Context,
    principal: str,
    action: str,
    scope: str,
    scope_id: str,
    reason: str,
) -> None:
    context.check_model = PermissionCheck(
        principal=principal,
        action=PermissionAction(action),
        scope=PermissionScope(scope),
        scope_id=scope_id,
        result=False,
        reason=reason,
    )


@then("the check result should be false")
def step_check_model_result_false(context: Context) -> None:
    assert context.check_model.result is False


# ── enforce_permission with pre-loaded service ────────────────────


@given(
    'a pre-loaded permission service with "{principal}" as "{role}" '
    'on "{scope}" scope "{scope_id}" for action "{action}"'
)
def step_given_decorated_fn_preloaded(
    context: Context,
    principal: str,
    role: str,
    scope: str,
    scope_id: str,
    action: str,
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    context.preloaded_service = PermissionService(bindings=[binding])
    context.preloaded_principal = principal
    context.enforce_action = action
    context.enforce_scope = scope
    context.enforce_scope_id = scope_id


@when("I call the pre-loaded decorated function in server mode")
def step_call_preloaded_decorated_server(context: Context) -> None:
    _set_server_mode()

    @enforce_permission(
        PermissionAction(context.enforce_action),
        PermissionScope(context.enforce_scope),
        context.enforce_scope_id,
        principal=context.preloaded_principal,
        service=context.preloaded_service,
    )
    def _guarded() -> str:
        return "ok"

    try:
        context.decorated_result = _guarded()
        context.decorated_error = None
    except PermissionError as exc:
        context.decorated_result = None
        context.decorated_error = exc


# ── enforce_permission with module-level default service ──────────


@given(
    'a module-level default service with "{principal}" as "{role}" '
    'on "{scope}" scope "{scope_id}"'
)
def step_given_module_default_service(
    context: Context,
    principal: str,
    role: str,
    scope: str,
    scope_id: str,
) -> None:
    binding = RoleBinding(
        principal=principal,
        role=PermissionRole(role),
        scope=PermissionScope(scope),
        scope_id=scope_id,
    )
    svc = PermissionService(bindings=[binding])
    set_default_permission_service(svc)
    context.default_svc_principal = principal


@when("I call the decorated function in server mode via module default")
def step_call_decorated_server_module_default(context: Context) -> None:
    _set_server_mode()

    @enforce_permission(
        PermissionAction(context.enforce_action),
        PermissionScope(context.enforce_scope),
        context.enforce_scope_id,
        principal=context.default_svc_principal,
    )
    def _guarded() -> str:
        return "ok"

    try:
        context.decorated_result = _guarded()
        context.decorated_error = None
    except PermissionError as exc:
        context.decorated_result = None
        context.decorated_error = exc
    finally:
        # Clean up module-level default
        set_default_permission_service(None)


# ── Input validation — check_permission ───────────────────────────


@when("I check permission with empty principal")
def step_check_permission_empty_principal(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.check_permission(
            principal="",
            action=PermissionAction.READ,
            scope=PermissionScope.PROJECT,
            scope_id="proj-1",
        )
    except ValueError as exc:
        context.value_error = exc


@when("I check permission with whitespace-only principal")
def step_check_permission_whitespace_principal(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.check_permission(
            principal="   ",
            action=PermissionAction.READ,
            scope=PermissionScope.PROJECT,
            scope_id="proj-1",
        )
    except ValueError as exc:
        context.value_error = exc


@when("I check permission with empty scope_id")
def step_check_permission_empty_scope_id(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.check_permission(
            principal="alice",
            action=PermissionAction.READ,
            scope=PermissionScope.PROJECT,
            scope_id="",
        )
    except ValueError as exc:
        context.value_error = exc


@when("I check permission with whitespace-only scope_id")
def step_check_permission_whitespace_scope_id(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.check_permission(
            principal="alice",
            action=PermissionAction.READ,
            scope=PermissionScope.PROJECT,
            scope_id="   ",
        )
    except ValueError as exc:
        context.value_error = exc


@then('a permission ValueError should mention "{fragment}"')
def step_permission_value_error_with_message(context: Context, fragment: str) -> None:
    assert context.value_error is not None, "Expected a ValueError"
    assert isinstance(context.value_error, ValueError)
    assert fragment in str(context.value_error), (
        f"Expected '{fragment}' in '{context.value_error}'"
    )


# ── Input validation — get_role_bindings ──────────────────────────


@when("I get role bindings with empty principal")
def step_get_bindings_empty_principal(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.get_role_bindings(
            principal="",
            scope=PermissionScope.PROJECT,
            scope_id="proj-1",
        )
    except ValueError as exc:
        context.value_error = exc


@when("I get role bindings with whitespace-only scope_id")
def step_get_bindings_whitespace_scope_id(context: Context) -> None:
    context.value_error = None
    try:
        context.permission_service.get_role_bindings(
            principal="alice",
            scope=PermissionScope.PROJECT,
            scope_id="   ",
        )
    except ValueError as exc:
        context.value_error = exc