356 lines
12 KiB
Python
356 lines
12 KiB
Python
|
"""Unit tests for P1-7 permission system fixes.
|
||
|
|
|
||
|
|
Tests:
|
||
|
|
1. _merge_field_permissions: strictest-wins merge logic
|
||
|
|
2. invalidate_all_user_permissions: SCAN-based (no KEYS)
|
||
|
|
3. get_cached_permissions: version validation logic
|
||
|
|
4. require_write: no broad wildcard *:write
|
||
|
|
"""
|
||
|
|
|
||
|
|
import asyncio
|
||
|
|
import json
|
||
|
|
import logging
|
||
|
|
import uuid
|
||
|
|
|
||
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
|
||
|
|
from app.core.permissions import (
|
||
|
|
_FIELD_PERM_SEVERITY,
|
||
|
|
_merge_field_permissions,
|
||
|
|
_matches_permission,
|
||
|
|
_normalize_permissions,
|
||
|
|
check_permission,
|
||
|
|
CACHE_PREFIX,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
class TestMergeFieldPermissions:
|
||
|
|
"""Tests for _merge_field_permissions — strictest-wins merge."""
|
||
|
|
|
||
|
|
def test_empty_incoming_no_change(self):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
_merge_field_permissions(existing, {})
|
||
|
|
assert existing == {"contacts": {"name": "read"}}
|
||
|
|
|
||
|
|
def test_new_module_added(self):
|
||
|
|
existing = {}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "hidden"}})
|
||
|
|
assert existing == {"contacts": {"name": "hidden"}}
|
||
|
|
|
||
|
|
def test_strictest_wins_hidden_over_read(self):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "hidden"}})
|
||
|
|
assert existing["contacts"]["name"] == "hidden"
|
||
|
|
|
||
|
|
def test_strictest_wins_read_does_not_override_hidden(self):
|
||
|
|
existing = {"contacts": {"name": "hidden"}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "read"}})
|
||
|
|
assert existing["contacts"]["name"] == "hidden"
|
||
|
|
|
||
|
|
def test_strictest_wins_readonly_over_read(self):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "readonly"}})
|
||
|
|
assert existing["contacts"]["name"] == "readonly"
|
||
|
|
|
||
|
|
def test_strictest_wins_readonly_does_not_override_hidden(self):
|
||
|
|
existing = {"contacts": {"name": "hidden"}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "readonly"}})
|
||
|
|
assert existing["contacts"]["name"] == "hidden"
|
||
|
|
|
||
|
|
def test_multiple_fields_merge_independently(self):
|
||
|
|
existing = {"contacts": {"name": "hidden", "email": "read"}}
|
||
|
|
_merge_field_permissions(
|
||
|
|
existing,
|
||
|
|
{"contacts": {"name": "read", "email": "hidden", "phone": "readonly"}},
|
||
|
|
)
|
||
|
|
assert existing["contacts"]["name"] == "hidden" # hidden stayed
|
||
|
|
assert existing["contacts"]["email"] == "hidden" # read upgraded to hidden
|
||
|
|
assert existing["contacts"]["phone"] == "readonly" # new field added
|
||
|
|
|
||
|
|
def test_multiple_modules_merge_independently(self):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
_merge_field_permissions(
|
||
|
|
existing,
|
||
|
|
{"users": {"email": "hidden"}, "contacts": {"name": "readonly"}},
|
||
|
|
)
|
||
|
|
assert existing["contacts"]["name"] == "readonly"
|
||
|
|
assert existing["users"]["email"] == "hidden"
|
||
|
|
|
||
|
|
def test_unknown_permission_level_skipped(self, caplog):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
with caplog.at_level(logging.WARNING):
|
||
|
|
_merge_field_permissions(
|
||
|
|
existing,
|
||
|
|
{"contacts": {"name": "bogus"}},
|
||
|
|
)
|
||
|
|
assert existing["contacts"]["name"] == "read" # unchanged
|
||
|
|
assert "Unknown field permission level" in caplog.text
|
||
|
|
|
||
|
|
def test_non_dict_fields_skipped(self):
|
||
|
|
existing = {}
|
||
|
|
_merge_field_permissions(existing, {"contacts": "not_a_dict"})
|
||
|
|
assert existing == {}
|
||
|
|
|
||
|
|
def test_non_string_perm_skipped(self):
|
||
|
|
existing = {"contacts": {}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": 123}})
|
||
|
|
assert existing["contacts"] == {}
|
||
|
|
|
||
|
|
def test_case_insensitive_perm_level(self):
|
||
|
|
existing = {"contacts": {"name": "read"}}
|
||
|
|
_merge_field_permissions(existing, {"contacts": {"name": "HIDDEN"}})
|
||
|
|
assert existing["contacts"]["name"] == "hidden"
|
||
|
|
|
||
|
|
def test_severity_ordering_constant(self):
|
||
|
|
assert _FIELD_PERM_SEVERITY["hidden"] > _FIELD_PERM_SEVERITY["readonly"]
|
||
|
|
assert _FIELD_PERM_SEVERITY["readonly"] > _FIELD_PERM_SEVERITY["read"]
|
||
|
|
|
||
|
|
|
||
|
|
class TestInvalidateAllUserPermissions:
|
||
|
|
"""Tests for invalidate_all_user_permissions — SCAN-based, no KEYS."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_scan_deletes_all_matching_keys(self):
|
||
|
|
from app.core.permissions import invalidate_all_user_permissions
|
||
|
|
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
# Simulate SCAN returning keys in two batches then finishing
|
||
|
|
call_count = 0
|
||
|
|
|
||
|
|
async def fake_scan(cursor, match, count):
|
||
|
|
nonlocal call_count
|
||
|
|
call_count += 1
|
||
|
|
if call_count == 1:
|
||
|
|
return (
|
||
|
|
1, # non-zero cursor = more to scan
|
||
|
|
[
|
||
|
|
f"{CACHE_PREFIX}:user1:{tenant_id}",
|
||
|
|
f"{CACHE_PREFIX}:user2:{tenant_id}",
|
||
|
|
],
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
return (
|
||
|
|
0, # done
|
||
|
|
[f"{CACHE_PREFIX}:user3:{tenant_id}"],
|
||
|
|
)
|
||
|
|
|
||
|
|
redis_mock.scan = fake_scan
|
||
|
|
redis_mock.delete = AsyncMock()
|
||
|
|
|
||
|
|
await invalidate_all_user_permissions(redis_mock, tenant_id)
|
||
|
|
|
||
|
|
# delete should be called twice — once per batch
|
||
|
|
assert redis_mock.delete.call_count == 2
|
||
|
|
# First batch: 2 keys
|
||
|
|
first_call_args = redis_mock.delete.call_args_list[0]
|
||
|
|
assert len(first_call_args[0]) == 2
|
||
|
|
# Second batch: 1 key
|
||
|
|
second_call_args = redis_mock.delete.call_args_list[1]
|
||
|
|
assert len(second_call_args[0]) == 1
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_scan_no_keys_no_delete(self):
|
||
|
|
from app.core.permissions import invalidate_all_user_permissions
|
||
|
|
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
|
||
|
|
async def fake_scan(cursor, match, count):
|
||
|
|
return (0, []) # no keys found
|
||
|
|
|
||
|
|
redis_mock.scan = fake_scan
|
||
|
|
redis_mock.delete = AsyncMock()
|
||
|
|
|
||
|
|
await invalidate_all_user_permissions(redis_mock, tenant_id)
|
||
|
|
|
||
|
|
# delete should not be called when no keys found
|
||
|
|
redis_mock.delete.assert_not_called()
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_scan_does_not_use_keys_command(self):
|
||
|
|
"""Ensure invalidate_all_user_permissions uses SCAN, not KEYS."""
|
||
|
|
from app.core.permissions import invalidate_all_user_permissions
|
||
|
|
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
|
||
|
|
async def fake_scan(cursor, match, count):
|
||
|
|
return (0, [])
|
||
|
|
|
||
|
|
redis_mock.scan = fake_scan
|
||
|
|
redis_mock.keys = AsyncMock()
|
||
|
|
|
||
|
|
await invalidate_all_user_permissions(redis_mock, tenant_id)
|
||
|
|
|
||
|
|
# keys() must never be called
|
||
|
|
redis_mock.keys.assert_not_called()
|
||
|
|
|
||
|
|
|
||
|
|
class TestGetCachedPermissionsVersionCheck:
|
||
|
|
"""Tests for get_cached_permissions version validation."""
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_cache_hit_version_match_returns_cached(self):
|
||
|
|
from app.core.permissions import get_cached_permissions
|
||
|
|
|
||
|
|
user_id = uuid.uuid4()
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
|
||
|
|
|
||
|
|
cached_data = {
|
||
|
|
"permissions": ["contacts:read"],
|
||
|
|
"denied": [],
|
||
|
|
"field_permissions": {},
|
||
|
|
"is_system_admin": False,
|
||
|
|
"version": 5,
|
||
|
|
}
|
||
|
|
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
redis_mock.get = AsyncMock(return_value=json.dumps(cached_data))
|
||
|
|
|
||
|
|
db_mock = AsyncMock()
|
||
|
|
|
||
|
|
# Mock _get_current_permission_version to return matching version
|
||
|
|
with patch(
|
||
|
|
"app.core.permissions._get_current_permission_version",
|
||
|
|
new_callable=AsyncMock,
|
||
|
|
return_value=5,
|
||
|
|
):
|
||
|
|
result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)
|
||
|
|
|
||
|
|
assert result == cached_data
|
||
|
|
redis_mock.setex.assert_not_called() # no re-caching needed
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_cache_version_mismatch_re_resolves(self):
|
||
|
|
from app.core.permissions import get_cached_permissions
|
||
|
|
|
||
|
|
user_id = uuid.uuid4()
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
cache_key = f"{CACHE_PREFIX}:{user_id}:{tenant_id}"
|
||
|
|
|
||
|
|
cached_data = {
|
||
|
|
"permissions": ["contacts:read"],
|
||
|
|
"denied": [],
|
||
|
|
"field_permissions": {},
|
||
|
|
"is_system_admin": False,
|
||
|
|
"version": 3, # stale version
|
||
|
|
}
|
||
|
|
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
redis_mock.get = AsyncMock(return_value=json.dumps(cached_data))
|
||
|
|
redis_mock.delete = AsyncMock()
|
||
|
|
redis_mock.setex = AsyncMock()
|
||
|
|
|
||
|
|
db_mock = AsyncMock()
|
||
|
|
|
||
|
|
resolved = {
|
||
|
|
"permissions": {"contacts:read", "contacts:write"},
|
||
|
|
"denied": set(),
|
||
|
|
"field_permissions": {},
|
||
|
|
"is_system_admin": False,
|
||
|
|
"version": 5, # new version
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch(
|
||
|
|
"app.core.permissions._get_current_permission_version",
|
||
|
|
new_callable=AsyncMock,
|
||
|
|
return_value=5, # current version differs from cached
|
||
|
|
), patch(
|
||
|
|
"app.core.permissions.resolve_permissions",
|
||
|
|
new_callable=AsyncMock,
|
||
|
|
return_value=resolved,
|
||
|
|
):
|
||
|
|
result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)
|
||
|
|
|
||
|
|
# Stale cache should be deleted
|
||
|
|
redis_mock.delete.assert_called_once_with(cache_key)
|
||
|
|
# New data should be cached
|
||
|
|
redis_mock.setex.assert_called_once()
|
||
|
|
# Result should have updated permissions
|
||
|
|
assert set(result["permissions"]) == {"contacts:read", "contacts:write"}
|
||
|
|
assert result["version"] == 5
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_cache_miss_resolves_from_db(self):
|
||
|
|
from app.core.permissions import get_cached_permissions
|
||
|
|
|
||
|
|
user_id = uuid.uuid4()
|
||
|
|
tenant_id = uuid.uuid4()
|
||
|
|
|
||
|
|
redis_mock = AsyncMock()
|
||
|
|
redis_mock.get = AsyncMock(return_value=None) # cache miss
|
||
|
|
redis_mock.setex = AsyncMock()
|
||
|
|
|
||
|
|
db_mock = AsyncMock()
|
||
|
|
|
||
|
|
resolved = {
|
||
|
|
"permissions": {"contacts:read"},
|
||
|
|
"denied": set(),
|
||
|
|
"field_permissions": {},
|
||
|
|
"is_system_admin": False,
|
||
|
|
"version": 1,
|
||
|
|
}
|
||
|
|
|
||
|
|
with patch(
|
||
|
|
"app.core.permissions.resolve_permissions",
|
||
|
|
new_callable=AsyncMock,
|
||
|
|
return_value=resolved,
|
||
|
|
):
|
||
|
|
result = await get_cached_permissions(db_mock, redis_mock, user_id, tenant_id)
|
||
|
|
|
||
|
|
assert set(result["permissions"]) == {"contacts:read"}
|
||
|
|
redis_mock.setex.assert_called_once()
|
||
|
|
|
||
|
|
|
||
|
|
class TestRequireWriteNoWildcard:
|
||
|
|
"""Tests that require_write does not use broad *:write wildcard."""
|
||
|
|
|
||
|
|
def test_write_permissions_list_has_no_wildcard(self):
|
||
|
|
from app.deps import _WRITE_PERMISSIONS
|
||
|
|
|
||
|
|
for perm in _WRITE_PERMISSIONS:
|
||
|
|
# No broad wildcards like *:write or *:create
|
||
|
|
assert not perm.startswith("*:"), f"Found wildcard permission: {perm}"
|
||
|
|
# All permissions should be module:action format
|
||
|
|
assert ":" in perm, f"Invalid permission format: {perm}"
|
||
|
|
|
||
|
|
def test_write_permissions_list_includes_contacts_write(self):
|
||
|
|
from app.deps import _WRITE_PERMISSIONS
|
||
|
|
|
||
|
|
assert "contacts:write" in _WRITE_PERMISSIONS
|
||
|
|
|
||
|
|
|
||
|
|
class TestCheckPermissionDenyList:
|
||
|
|
"""Verify deny list still works correctly."""
|
||
|
|
|
||
|
|
def test_deny_overrides_allowed(self):
|
||
|
|
resolved = {
|
||
|
|
"permissions": {"contacts:read", "contacts:write"},
|
||
|
|
"denied": {"contacts:write"},
|
||
|
|
"is_system_admin": False,
|
||
|
|
}
|
||
|
|
assert check_permission(resolved, "contacts:read") is True
|
||
|
|
assert check_permission(resolved, "contacts:write") is False
|
||
|
|
|
||
|
|
def test_deny_wildcard_blocks_specific(self):
|
||
|
|
resolved = {
|
||
|
|
"permissions": {"contacts:read"},
|
||
|
|
"denied": {"contacts:*"},
|
||
|
|
"is_system_admin": False,
|
||
|
|
}
|
||
|
|
assert check_permission(resolved, "contacts:read") is False
|
||
|
|
assert check_permission(resolved, "contacts:write") is False
|
||
|
|
|
||
|
|
def test_system_admin_ignores_deny(self):
|
||
|
|
resolved = {
|
||
|
|
"permissions": set(),
|
||
|
|
"denied": {"contacts:*"},
|
||
|
|
"is_system_admin": True,
|
||
|
|
}
|
||
|
|
assert check_permission(resolved, "contacts:read") is True
|