cardosofelipe/syndarix
1
0
Fork 0
forked from cardosofelipe/fast-next-template
Code Issues 18 Pull Requests Wiki Activity
Files
60ebeaa582b6db4a05f07a3593de66993929c104
syndarix/backend/tests/services/safety/test_permissions.py
Felipe Cardoso 60ebeaa582 test(safety): add comprehensive tests for safety framework modules 
Add tests to improve backend coverage from 85% to 93%:

- test_audit.py: 60 tests for AuditLogger (20% -> 99%)
  - Hash chain integrity, sanitization, retention, handlers
  - Fixed bug: hash chain modification after event creation
  - Fixed bug: verification not using correct prev_hash

- test_hitl.py: Tests for HITL manager (0% -> 100%)
- test_permissions.py: Tests for permissions manager (0% -> 99%)
- test_rollback.py: Tests for rollback manager (0% -> 100%)
- test_metrics.py: Tests for metrics collector (0% -> 100%)
- test_mcp_integration.py: Tests for MCP safety wrapper (0% -> 100%)
- test_validation.py: Additional cache and edge case tests (76% -> 100%)
- test_scoring.py: Lock cleanup and edge case tests (78% -> 91%)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-01-04 19:41:54 +01:00

934 lines
30 KiB
Python
Raw Blame History

"""Tests for Permission Manager.
Tests cover:
- PermissionGrant: creation, expiry, matching, hierarchy
- PermissionManager: grant, revoke, check, require, list, defaults
- Edge cases: wildcards, expiration, default deny/allow
"""
from datetime import datetime, timedelta
import pytest
import pytest_asyncio
from app.services.safety.exceptions import PermissionDeniedError
from app.services.safety.models import (
ActionMetadata,
ActionRequest,
ActionType,
PermissionLevel,
ResourceType,
)
from app.services.safety.permissions.manager import PermissionGrant, PermissionManager
# ============================================================================
# Fixtures
# ============================================================================
@pytest.fixture
def action_metadata() -> ActionMetadata:
"""Create standard action metadata for tests."""
return ActionMetadata(
agent_id="test-agent",
project_id="test-project",
session_id="test-session",
)
@pytest_asyncio.fixture
async def permission_manager() -> PermissionManager:
"""Create a PermissionManager for testing."""
return PermissionManager(default_deny=True)
@pytest_asyncio.fixture
async def permissive_manager() -> PermissionManager:
"""Create a PermissionManager with default_deny=False."""
return PermissionManager(default_deny=False)
# ============================================================================
# PermissionGrant Tests
# ============================================================================
class TestPermissionGrant:
"""Tests for the PermissionGrant class."""
def test_grant_creation(self) -> None:
"""Test basic grant creation."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
granted_by="admin",
reason="Read access to data directory",
)
assert grant.id is not None
assert grant.agent_id == "agent-1"
assert grant.resource_pattern == "/data/*"
assert grant.resource_type == ResourceType.FILE
assert grant.level == PermissionLevel.READ
assert grant.granted_by == "admin"
assert grant.reason == "Read access to data directory"
assert grant.expires_at is None
assert grant.created_at is not None
def test_grant_with_expiration(self) -> None:
"""Test grant with expiration time."""
future = datetime.utcnow() + timedelta(hours=1)
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
expires_at=future,
)
assert grant.expires_at == future
assert grant.is_expired() is False
def test_is_expired_no_expiration(self) -> None:
"""Test is_expired with no expiration set."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
assert grant.is_expired() is False
def test_is_expired_future(self) -> None:
"""Test is_expired with future expiration."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
expires_at=datetime.utcnow() + timedelta(hours=1),
)
assert grant.is_expired() is False
def test_is_expired_past(self) -> None:
"""Test is_expired with past expiration."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
expires_at=datetime.utcnow() - timedelta(hours=1),
)
assert grant.is_expired() is True
def test_matches_exact(self) -> None:
"""Test matching with exact pattern."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="/data/file.txt",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
assert grant.matches("/data/file.txt", ResourceType.FILE) is True
assert grant.matches("/data/other.txt", ResourceType.FILE) is False
def test_matches_wildcard(self) -> None:
"""Test matching with wildcard pattern."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
assert grant.matches("/data/file.txt", ResourceType.FILE) is True
# fnmatch's * matches everything including /
assert grant.matches("/data/subdir/file.txt", ResourceType.FILE) is True
assert grant.matches("/other/file.txt", ResourceType.FILE) is False
def test_matches_recursive_wildcard(self) -> None:
"""Test matching with recursive pattern."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="/data/**",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# fnmatch treats ** similar to * - both match everything including /
assert grant.matches("/data/file.txt", ResourceType.FILE) is True
assert grant.matches("/data/subdir/file.txt", ResourceType.FILE) is True
def test_matches_wrong_resource_type(self) -> None:
"""Test matching fails with wrong resource type."""
grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Same pattern but different resource type
assert grant.matches("/data/table", ResourceType.DATABASE) is False
def test_allows_hierarchy(self) -> None:
"""Test permission level hierarchy."""
admin_grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.ADMIN,
)
# ADMIN allows all levels
assert admin_grant.allows(PermissionLevel.NONE) is True
assert admin_grant.allows(PermissionLevel.READ) is True
assert admin_grant.allows(PermissionLevel.WRITE) is True
assert admin_grant.allows(PermissionLevel.EXECUTE) is True
assert admin_grant.allows(PermissionLevel.DELETE) is True
assert admin_grant.allows(PermissionLevel.ADMIN) is True
def test_allows_read_only(self) -> None:
"""Test READ grant only allows READ and NONE."""
read_grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
assert read_grant.allows(PermissionLevel.NONE) is True
assert read_grant.allows(PermissionLevel.READ) is True
assert read_grant.allows(PermissionLevel.WRITE) is False
assert read_grant.allows(PermissionLevel.EXECUTE) is False
assert read_grant.allows(PermissionLevel.DELETE) is False
assert read_grant.allows(PermissionLevel.ADMIN) is False
def test_allows_write_includes_read(self) -> None:
"""Test WRITE grant includes READ."""
write_grant = PermissionGrant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.WRITE,
)
assert write_grant.allows(PermissionLevel.READ) is True
assert write_grant.allows(PermissionLevel.WRITE) is True
assert write_grant.allows(PermissionLevel.EXECUTE) is False
# ============================================================================
# PermissionManager Tests
# ============================================================================
class TestPermissionManager:
"""Tests for the PermissionManager class."""
@pytest.mark.asyncio
async def test_grant_creates_permission(
self,
permission_manager: PermissionManager,
) -> None:
"""Test granting a permission."""
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
granted_by="admin",
reason="Read access",
)
assert grant.id is not None
assert grant.agent_id == "agent-1"
assert grant.resource_pattern == "/data/*"
@pytest.mark.asyncio
async def test_grant_with_duration(
self,
permission_manager: PermissionManager,
) -> None:
"""Test granting a temporary permission."""
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
duration_seconds=3600, # 1 hour
)
assert grant.expires_at is not None
assert grant.is_expired() is False
@pytest.mark.asyncio
async def test_revoke_by_id(
self,
permission_manager: PermissionManager,
) -> None:
"""Test revoking a grant by ID."""
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
success = await permission_manager.revoke(grant.id)
assert success is True
# Verify grant is removed
grants = await permission_manager.list_grants(agent_id="agent-1")
assert len(grants) == 0
@pytest.mark.asyncio
async def test_revoke_nonexistent(
self,
permission_manager: PermissionManager,
) -> None:
"""Test revoking a non-existent grant."""
success = await permission_manager.revoke("nonexistent-id")
assert success is False
@pytest.mark.asyncio
async def test_revoke_all_for_agent(
self,
permission_manager: PermissionManager,
) -> None:
"""Test revoking all permissions for an agent."""
# Grant multiple permissions
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/api/*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
)
await permission_manager.grant(
agent_id="agent-2",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
revoked = await permission_manager.revoke_all("agent-1")
assert revoked == 2
# Verify agent-1 grants are gone
grants = await permission_manager.list_grants(agent_id="agent-1")
assert len(grants) == 0
# Verify agent-2 grant remains
grants = await permission_manager.list_grants(agent_id="agent-2")
assert len(grants) == 1
@pytest.mark.asyncio
async def test_revoke_all_no_grants(
self,
permission_manager: PermissionManager,
) -> None:
"""Test revoking all when no grants exist."""
revoked = await permission_manager.revoke_all("nonexistent-agent")
assert revoked == 0
@pytest.mark.asyncio
async def test_check_granted(
self,
permission_manager: PermissionManager,
) -> None:
"""Test checking a granted permission."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert allowed is True
@pytest.mark.asyncio
async def test_check_denied_default_deny(
self,
permission_manager: PermissionManager,
) -> None:
"""Test checking denied with default_deny=True."""
# No grants, should be denied
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert allowed is False
@pytest.mark.asyncio
async def test_check_uses_default_permissions(
self,
permissive_manager: PermissionManager,
) -> None:
"""Test that default permissions apply when default_deny=False."""
# No explicit grants, but FILE default is READ
allowed = await permissive_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert allowed is True
# But WRITE should fail
allowed = await permissive_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.WRITE,
)
assert allowed is False
@pytest.mark.asyncio
async def test_check_shell_denied_by_default(
self,
permissive_manager: PermissionManager,
) -> None:
"""Test SHELL is denied by default (NONE level)."""
allowed = await permissive_manager.check(
agent_id="agent-1",
resource="rm -rf /",
resource_type=ResourceType.SHELL,
required_level=PermissionLevel.EXECUTE,
)
assert allowed is False
@pytest.mark.asyncio
async def test_check_expired_grant_ignored(
self,
permission_manager: PermissionManager,
) -> None:
"""Test that expired grants are ignored in checks."""
# Create an already-expired grant
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
duration_seconds=1, # Very short
)
# Manually expire it
grant.expires_at = datetime.utcnow() - timedelta(seconds=10)
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert allowed is False
@pytest.mark.asyncio
async def test_check_insufficient_level(
self,
permission_manager: PermissionManager,
) -> None:
"""Test check fails when grant level is insufficient."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Try to get WRITE access with only READ grant
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.WRITE,
)
assert allowed is False
@pytest.mark.asyncio
async def test_check_action_file_read(
self,
permission_manager: PermissionManager,
action_metadata: ActionMetadata,
) -> None:
"""Test check_action for file read."""
await permission_manager.grant(
agent_id="test-agent",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
action = ActionRequest(
action_type=ActionType.FILE_READ,
resource="/data/file.txt",
metadata=action_metadata,
)
allowed = await permission_manager.check_action(action)
assert allowed is True
@pytest.mark.asyncio
async def test_check_action_file_write(
self,
permission_manager: PermissionManager,
action_metadata: ActionMetadata,
) -> None:
"""Test check_action for file write."""
await permission_manager.grant(
agent_id="test-agent",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.WRITE,
)
action = ActionRequest(
action_type=ActionType.FILE_WRITE,
resource="/data/file.txt",
metadata=action_metadata,
)
allowed = await permission_manager.check_action(action)
assert allowed is True
@pytest.mark.asyncio
async def test_check_action_uses_tool_name_as_resource(
self,
permission_manager: PermissionManager,
action_metadata: ActionMetadata,
) -> None:
"""Test check_action uses tool_name when resource is None."""
await permission_manager.grant(
agent_id="test-agent",
resource_pattern="search_*",
resource_type=ResourceType.CUSTOM,
level=PermissionLevel.EXECUTE,
)
action = ActionRequest(
action_type=ActionType.TOOL_CALL,
tool_name="search_documents",
resource=None,
metadata=action_metadata,
)
allowed = await permission_manager.check_action(action)
assert allowed is True
@pytest.mark.asyncio
async def test_require_permission_granted(
self,
permission_manager: PermissionManager,
) -> None:
"""Test require_permission doesn't raise when granted."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Should not raise
await permission_manager.require_permission(
agent_id="agent-1",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
@pytest.mark.asyncio
async def test_require_permission_denied(
self,
permission_manager: PermissionManager,
) -> None:
"""Test require_permission raises when denied."""
with pytest.raises(PermissionDeniedError) as exc_info:
await permission_manager.require_permission(
agent_id="agent-1",
resource="/secret/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert "/secret/file.txt" in str(exc_info.value)
assert exc_info.value.agent_id == "agent-1"
assert exc_info.value.required_permission == "read"
@pytest.mark.asyncio
async def test_list_grants_all(
self,
permission_manager: PermissionManager,
) -> None:
"""Test listing all grants."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
await permission_manager.grant(
agent_id="agent-2",
resource_pattern="/api/*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
)
grants = await permission_manager.list_grants()
assert len(grants) == 2
@pytest.mark.asyncio
async def test_list_grants_by_agent(
self,
permission_manager: PermissionManager,
) -> None:
"""Test listing grants filtered by agent."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
await permission_manager.grant(
agent_id="agent-2",
resource_pattern="/api/*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
)
grants = await permission_manager.list_grants(agent_id="agent-1")
assert len(grants) == 1
assert grants[0].agent_id == "agent-1"
@pytest.mark.asyncio
async def test_list_grants_by_resource_type(
self,
permission_manager: PermissionManager,
) -> None:
"""Test listing grants filtered by resource type."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/api/*",
resource_type=ResourceType.API,
level=PermissionLevel.EXECUTE,
)
grants = await permission_manager.list_grants(resource_type=ResourceType.FILE)
assert len(grants) == 1
assert grants[0].resource_type == ResourceType.FILE
@pytest.mark.asyncio
async def test_list_grants_excludes_expired(
self,
permission_manager: PermissionManager,
) -> None:
"""Test that list_grants excludes expired grants."""
# Create expired grant
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/old/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
duration_seconds=1,
)
grant.expires_at = datetime.utcnow() - timedelta(seconds=10)
# Create valid grant
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/new/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
grants = await permission_manager.list_grants()
assert len(grants) == 1
assert grants[0].resource_pattern == "/new/*"
def test_set_default_permission(
self,
) -> None:
"""Test setting default permission level."""
manager = PermissionManager(default_deny=False)
# Default for SHELL is NONE
assert manager._default_permissions[ResourceType.SHELL] == PermissionLevel.NONE
# Change it
manager.set_default_permission(ResourceType.SHELL, PermissionLevel.EXECUTE)
assert (
manager._default_permissions[ResourceType.SHELL] == PermissionLevel.EXECUTE
)
@pytest.mark.asyncio
async def test_set_default_permission_affects_checks(
self,
permissive_manager: PermissionManager,
) -> None:
"""Test that changing default permissions affects checks."""
# Initially SHELL is NONE
allowed = await permissive_manager.check(
agent_id="agent-1",
resource="ls",
resource_type=ResourceType.SHELL,
required_level=PermissionLevel.EXECUTE,
)
assert allowed is False
# Change default
permissive_manager.set_default_permission(
ResourceType.SHELL, PermissionLevel.EXECUTE
)
# Now should be allowed
allowed = await permissive_manager.check(
agent_id="agent-1",
resource="ls",
resource_type=ResourceType.SHELL,
required_level=PermissionLevel.EXECUTE,
)
assert allowed is True
# ============================================================================
# Edge Cases
# ============================================================================
class TestPermissionEdgeCases:
"""Edge cases that could reveal hidden bugs."""
@pytest.mark.asyncio
async def test_multiple_matching_grants(
self,
permission_manager: PermissionManager,
) -> None:
"""Test when multiple grants match - first sufficient one wins."""
# Grant READ on all files
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Also grant WRITE on specific path
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/writable/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.WRITE,
)
# Write on writable path should work
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/data/writable/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.WRITE,
)
assert allowed is True
@pytest.mark.asyncio
async def test_wildcard_all_pattern(
self,
permission_manager: PermissionManager,
) -> None:
"""Test * pattern matches everything."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="*",
resource_type=ResourceType.FILE,
level=PermissionLevel.ADMIN,
)
allowed = await permission_manager.check(
agent_id="agent-1",
resource="/any/path/anywhere/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.DELETE,
)
# fnmatch's * matches everything including /
assert allowed is True
@pytest.mark.asyncio
async def test_question_mark_wildcard(
self,
permission_manager: PermissionManager,
) -> None:
"""Test ? wildcard matches single character."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="file?.txt",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
assert (
await permission_manager.check(
agent_id="agent-1",
resource="file1.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
is True
)
assert (
await permission_manager.check(
agent_id="agent-1",
resource="file10.txt", # Two characters, won't match
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
is False
)
@pytest.mark.asyncio
async def test_concurrent_grant_revoke(
self,
permission_manager: PermissionManager,
) -> None:
"""Test concurrent grant and revoke operations."""
async def grant_many():
grants = []
for i in range(10):
g = await permission_manager.grant(
agent_id="agent-1",
resource_pattern=f"/path{i}/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
grants.append(g)
return grants
async def revoke_many(grants):
for g in grants:
await permission_manager.revoke(g.id)
grants = await grant_many()
await revoke_many(grants)
# All should be revoked
remaining = await permission_manager.list_grants(agent_id="agent-1")
assert len(remaining) == 0
@pytest.mark.asyncio
async def test_check_action_with_no_resource_or_tool(
self,
permission_manager: PermissionManager,
action_metadata: ActionMetadata,
) -> None:
"""Test check_action when both resource and tool_name are None."""
await permission_manager.grant(
agent_id="test-agent",
resource_pattern="*",
resource_type=ResourceType.LLM,
level=PermissionLevel.EXECUTE,
)
action = ActionRequest(
action_type=ActionType.LLM_CALL,
resource=None,
tool_name=None,
metadata=action_metadata,
)
# Should use "*" as fallback
allowed = await permission_manager.check_action(action)
assert allowed is True
@pytest.mark.asyncio
async def test_cleanup_expired_called_on_check(
self,
permission_manager: PermissionManager,
) -> None:
"""Test that expired grants are cleaned up during check."""
# Create expired grant
grant = await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/old/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
duration_seconds=1,
)
grant.expires_at = datetime.utcnow() - timedelta(seconds=10)
# Create valid grant
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/new/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Run a check - this should trigger cleanup
await permission_manager.check(
agent_id="agent-1",
resource="/new/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
# Now verify expired grant was cleaned up
async with permission_manager._lock:
assert len(permission_manager._grants) == 1
assert permission_manager._grants[0].resource_pattern == "/new/*"
@pytest.mark.asyncio
async def test_check_wrong_agent_id(
self,
permission_manager: PermissionManager,
) -> None:
"""Test check fails for different agent."""
await permission_manager.grant(
agent_id="agent-1",
resource_pattern="/data/*",
resource_type=ResourceType.FILE,
level=PermissionLevel.READ,
)
# Different agent should not have access
allowed = await permission_manager.check(
agent_id="agent-2",
resource="/data/file.txt",
resource_type=ResourceType.FILE,
required_level=PermissionLevel.READ,
)
assert allowed is False
Reference in New Issue View Git Blame Copy Permalink