Leopoldadmin/leocrm
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
2a80aeb0af51b543da842c7d3a133696e0b0cc87
leocrm/tests/test_rbac_comprehensive.py
T

1877 lines
74 KiB
Python
Raw Normal View History

test: 102 comprehensive RBAC tests + fix: filter soft-deleted groups in permission resolver
2026-07-16 01:58:22 +02:00
"""Comprehensive RBAC system tests — covers all 5 RBAC phases.
Sections:
1. Permission-Resolver (app/core/permissions.py) — unit tests, no DB
2. Permission-Registry (app/core/permission_registry.py) — unit tests, no DB
3. Group CRUD (app/routes/groups.py + app/services/group_service.py) — DB required
4. RBAC Route-Guard (app/deps.py) — DB required
5. Self-Modification Prevention (app/routes/users.py) — DB required
6. Cache-Invalidierung (app/core/permissions.py) — DB + Redis required
7. Mail-Plugin RBAC (app/plugins/builtins/mail/routes.py) — DB required
8. Field-Level Permission (app/core/permissions.py + services) — unit + DB
9. Integration Tests — DB + Redis required
"""
from __future__ import annotations
import json
import uuid
from datetime import UTC, datetime
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
from app.core.auth import hash_password
from app.core.db import close_engine, reset_engine_for_testing
from app.core.permission_registry import (
CORE_PERMISSIONS,
PermissionRegistry,
get_permission_registry,
init_permission_registry,
)
from app.core.permissions import (
_matches_permission,
_normalize_permissions,
check_field_access,
check_permission,
filter_fields_by_permission,
invalidate_all_user_permissions,
invalidate_permission_cache,
resolve_permissions,
)
from app.core.service_container import get_container
from app.main import create_app
from app.models.group import Group, UserGroup
from app.models.role import Role
from app.models.user import User, UserTenant
from app.plugins.builtins.mail import MailPlugin
from app.plugins.builtins.mail.models import (
Mail,
MailAccount,
MailAccountDelegate,
MailFolder,
)
from app.plugins.registry import reset_registry_for_testing
from app.services.plugin_service import reset_plugin_service_for_testing
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
# ═══════════════════════════════════════════════════════════════
# CSRF-AWARE HELPERS
# ═══════════════════════════════════════════════════════════════
async def login_with_csrf(client: AsyncClient, email: str, password: str = "TestPass123!") -> str:
"""Login via HTTP API and return the CSRF token.
The client cookies are set automatically by httpx.
The CSRF token must be sent as X-CSRF-Token header on unsafe methods.
"""
resp = await client.post(
"/api/v1/auth/login",
json={"email": email, "password": password},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200, f"Login failed: {resp.status_code} {resp.text}"
return resp.json()["csrf_token"]
def csrf_headers(csrf_token: str) -> dict:
"""Return headers dict with Origin + X-CSRF-Token for unsafe methods."""
return {**ORIGIN_HEADER, "X-CSRF-Token": csrf_token}
# ═══════════════════════════════════════════════════════════════
# SECTION 1: Permission-Resolver Tests (unit tests, no DB needed)
# ═══════════════════════════════════════════════════════════════
class TestPermissionResolverUnit:
"""Unit tests for permission resolver functions — no DB required."""
# ── _matches_permission ──
def test_matches_exact_permission(self):
"""Exact match: companies:read matches companies:read."""
assert _matches_permission("companies:read", "companies:read") is True
def test_matches_wildcard_all(self):
"""Wildcard *:* matches everything."""
assert _matches_permission("*:*", "companies:read") is True
assert _matches_permission("*:*", "contacts:write") is True
assert _matches_permission("*:*", "mail:send") is True
def test_matches_wildcard_module(self):
"""Wildcard companies:* matches all actions for companies."""
assert _matches_permission("companies:*", "companies:read") is True
assert _matches_permission("companies:*", "companies:write") is True
assert _matches_permission("companies:*", "companies:delete") is True
def test_matches_wildcard_action(self):
"""Wildcard *:read matches read action for any module."""
assert _matches_permission("*:read", "companies:read") is True
assert _matches_permission("*:read", "contacts:read") is True
assert _matches_permission("*:read", "mail:read") is True
def test_matches_no_match_different_module(self):
"""companies:read does not match contacts:read."""
assert _matches_permission("companies:read", "contacts:read") is False
def test_matches_no_match_different_action(self):
"""companies:read does not match companies:write."""
assert _matches_permission("companies:read", "companies:write") is False
def test_matches_wildcard_module_does_not_match_other_module(self):
"""companies:* does not match contacts:read."""
assert _matches_permission("companies:*", "contacts:read") is False
def test_matches_wildcard_action_does_not_match_other_action(self):
"""*:read does not match companies:write."""
assert _matches_permission("*:read", "companies:write") is False
# ── _normalize_permissions ──
def test_normalize_list_format(self):
"""List format: ['companies:read', 'contacts:write'] → set."""
result = _normalize_permissions(["companies:read", "contacts:write"])
assert result == {"companies:read", "contacts:write"}
def test_normalize_dict_bool_format(self):
"""Dict-bool format: {'companies:read': true, 'contacts:write': false}."""
result = _normalize_permissions({
"companies:read": True,
"contacts:write": False,
})
assert result == {"companies:read"}
def test_normalize_dict_nested_format(self):
"""Dict-nested format: {'companies': {'read': true, 'write': false}}."""
result = _normalize_permissions({
"companies": {"read": True, "write": False, "delete": True},
"contacts": {"read": True},
})
assert result == {"companies:read", "companies:delete", "contacts:read"}
def test_normalize_empty(self):
"""Empty list/dict returns empty set."""
assert _normalize_permissions([]) == set()
assert _normalize_permissions({}) == set()
def test_normalize_dot_to_colon(self):
"""Dots are replaced with colons: 'companies.read' → 'companies:read'."""
result = _normalize_permissions(["companies.read"])
assert result == {"companies:read"}
# ── check_permission ──
def test_check_permission_system_admin_bypass(self):
"""System admin bypasses all permission checks."""
resolved = {"is_system_admin": True, "permissions": [], "denied": []}
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "anything:anything") is True
def test_check_permission_exact_match(self):
"""Exact permission match grants access."""
resolved = {"is_system_admin": False, "permissions": ["companies:read"], "denied": []}
assert check_permission(resolved, "companies:read") is True
def test_check_permission_wildcard_match(self):
"""Wildcard permissions match specific requirements."""
resolved = {"is_system_admin": False, "permissions": ["companies:*"], "denied": []}
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "companies:write") is True
def test_check_permission_denied_overrides_allowed(self):
"""Denied permission overrides allowed permission."""
resolved = {
"is_system_admin": False,
"permissions": ["companies:read", "companies:write"],
"denied": ["companies:write"],
}
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "companies:write") is False
def test_check_permission_denied_wildcard_overrides(self):
"""Denied wildcard overrides specific allowed permission."""
resolved = {
"is_system_admin": False,
"permissions": ["companies:read"],
"denied": ["companies:*"],
}
assert check_permission(resolved, "companies:read") is False
def test_check_permission_no_match(self):
"""No matching permission denies access."""
resolved = {"is_system_admin": False, "permissions": ["contacts:read"], "denied": []}
assert check_permission(resolved, "companies:read") is False
def test_check_permission_star_star_grants_all(self):
"""*:* grants any permission."""
resolved = {"is_system_admin": False, "permissions": ["*:*"], "denied": []}
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "mail:send") is True
# ── check_field_access ──
def test_check_field_access_hidden(self):
"""Hidden field returns 'hidden'."""
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
}
assert check_field_access(resolved, "companies", "annual_revenue") == "hidden"
def test_check_field_access_readonly(self):
"""Readonly field returns 'readonly'."""
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"name": "readonly"}},
}
assert check_field_access(resolved, "companies", "name") == "readonly"
def test_check_field_access_read_default(self):
"""Field without explicit permission returns default 'read'."""
resolved = {"is_system_admin": False, "field_permissions": {}}
assert check_field_access(resolved, "companies", "name") == "read"
def test_check_field_access_system_admin(self):
"""System admin always gets 'read' for any field."""
resolved = {
"is_system_admin": True,
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
}
assert check_field_access(resolved, "companies", "annual_revenue") == "read"
# ── filter_fields_by_permission ──
def test_filter_fields_removes_hidden(self):
"""Hidden fields are removed from the output."""
data = {"name": "ACME", "annual_revenue": 1000000, "industry": "IT"}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "annual_revenue" not in result
assert result["name"] == "ACME"
assert result["industry"] == "IT"
def test_filter_fields_keeps_readonly(self):
"""Readonly fields are kept in the output."""
data = {"name": "ACME", "industry": "IT"}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"name": "readonly"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "name" in result
assert result["name"] == "ACME"
def test_filter_fields_keeps_read(self):
"""Read fields are kept in the output."""
data = {"name": "ACME", "industry": "IT"}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"name": "read"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "name" in result
def test_filter_fields_system_admin_no_filtering(self):
"""System admin: no fields are filtered."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {
"is_system_admin": True,
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert result == data
def test_filter_fields_no_field_permissions_no_filtering(self):
"""No field_permissions in resolved → no filtering."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {"is_system_admin": False, "field_permissions": {}}
result = filter_fields_by_permission(data, resolved, "companies")
assert result == data
def test_filter_fields_no_module_perms_no_filtering(self):
"""No permissions for the specific module → no filtering."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {
"is_system_admin": False,
"field_permissions": {"contacts": {"email": "hidden"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert result == data
# ═══════════════════════════════════════════════════════════════
# SECTION 2: Permission-Registry Tests (unit tests, no DB needed)
# ═══════════════════════════════════════════════════════════════
_CORE_ONLY_COUNT = sum(1 for p in CORE_PERMISSIONS if p.get("category") == "core")
class TestPermissionRegistryUnit:
"""Unit tests for the PermissionRegistry — no DB required."""
def test_initialize_registers_core_permissions(self):
"""initialize() registers all CORE_PERMISSIONS."""
reg = PermissionRegistry()
reg.initialize()
for perm in CORE_PERMISSIONS:
assert reg.is_valid(perm["key"]), f"Missing core permission: {perm['key']}"
def test_initialize_with_active_plugins(self):
"""initialize() accepts active_plugin_names set."""
reg = PermissionRegistry()
reg.initialize(active_plugin_names={"mail", "dms"})
assert reg.is_plugin_active("mail") is True
assert reg.is_plugin_active("dms") is True
assert reg.is_plugin_active("calendar") is False
def test_register_plugin_permissions(self):
"""register_plugin_permissions adds plugin permissions."""
reg = PermissionRegistry()
reg.initialize()
reg.register_plugin_permissions("mail", ["mail:read", "mail:send", "mail:config"])
assert reg.is_valid("mail:read") is True
assert reg.is_valid("mail:send") is True
assert reg.is_valid("mail:config") is True
def test_unregister_plugin_permissions(self):
"""unregister_plugin_permissions removes plugin permissions."""
reg = PermissionRegistry()
reg.initialize()
reg.register_plugin_permissions("mail", ["mail:read", "mail:send"])
assert reg.is_valid("mail:read") is True
reg.unregister_plugin_permissions("mail")
assert reg.is_valid("mail:read") is False
assert reg.is_valid("mail:send") is False
def test_is_valid_unknown_permission(self):
"""is_valid returns False for unknown permission."""
reg = PermissionRegistry()
reg.initialize()
assert reg.is_valid("nonexistent:action") is False
def test_get_all_returns_all_permissions(self):
"""get_all() returns all registered permissions."""
reg = PermissionRegistry()
reg.initialize()
all_perms = reg.get_all()
assert len(all_perms) == len(CORE_PERMISSIONS)
reg.register_plugin_permissions("mail", ["mail:read"])
all_perms = reg.get_all()
assert len(all_perms) == len(CORE_PERMISSIONS) + 1
def test_get_core_returns_only_core(self):
"""get_core() returns only core-category permissions (excludes system:admin)."""
reg = PermissionRegistry()
reg.initialize()
reg.register_plugin_permissions("mail", ["mail:read"])
core = reg.get_core()
assert all(p.get("category") == "core" for p in core)
assert len(core) == _CORE_ONLY_COUNT
def test_get_plugin_permissions_returns_only_plugin(self):
"""get_plugin_permissions() returns only plugin permissions."""
reg = PermissionRegistry()
reg.initialize()
reg.register_plugin_permissions("mail", ["mail:read", "mail:send"])
plugin_perms = reg.get_plugin_permissions()
assert all(p.get("category") == "plugins" for p in plugin_perms)
assert len(plugin_perms) == 2
def test_get_grouped(self):
"""get_grouped() returns permissions grouped by category."""
reg = PermissionRegistry()
reg.initialize()
reg.register_plugin_permissions("mail", ["mail:read"])
grouped = reg.get_grouped()
assert "core" in grouped
assert "plugins" in grouped
assert len(grouped["core"]) == _CORE_ONLY_COUNT
assert len(grouped["plugins"]) == 1
def test_register_field_definitions(self):
"""register_field_definitions stores field defs for a plugin."""
reg = PermissionRegistry()
reg.initialize()
field_defs = [
{"module": "mail", "field": "subject", "label": "Subject", "sensitivity": "normal"},
]
reg.register_field_definitions("mail", field_defs)
all_defs = reg.get_all_field_definitions()
mail_defs = [d for d in all_defs if d.get("module") == "mail"]
assert len(mail_defs) == 1
assert mail_defs[0]["field"] == "subject"
def test_get_all_field_definitions_includes_core(self):
"""get_all_field_definitions() includes core field definitions."""
reg = PermissionRegistry()
reg.initialize()
all_defs = reg.get_all_field_definitions()
company_defs = [d for d in all_defs if d.get("module") == "companies"]
assert len(company_defs) > 0
# ═══════════════════════════════════════════════════════════════
# SECTION 6: Cache-Invalidierung Tests (DB + Redis required)
# ═══════════════════════════════════════════════════════════════
class TestPermissionCacheInvalidation:
"""Tests for Redis-based permission caching and invalidation."""
@pytest.mark.asyncio
async def test_get_cached_permissions_caches_after_first_call(
self, db_session: AsyncSession, redis_client
):
"""get_cached_permissions stores result in Redis after first call."""
from app.core.permissions import CACHE_PREFIX, get_cached_permissions
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
tenant = seed["tenant_a"]
result1 = await get_cached_permissions(db_session, redis_client, admin.id, tenant.id)
cache_key = f"{CACHE_PREFIX}:{admin.id}:{tenant.id}"
cached_raw = await redis_client.get(cache_key)
assert cached_raw is not None, "Cache should be populated after first call"
@pytest.mark.asyncio
async def test_invalidate_permission_cache_clears_user(
self, db_session: AsyncSession, redis_client
):
"""invalidate_permission_cache removes cache for a specific user+tenant."""
from app.core.permissions import CACHE_PREFIX, get_cached_permissions
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
tenant = seed["tenant_a"]
await get_cached_permissions(db_session, redis_client, admin.id, tenant.id)
cache_key = f"{CACHE_PREFIX}:{admin.id}:{tenant.id}"
assert await redis_client.get(cache_key) is not None
await invalidate_permission_cache(redis_client, admin.id, tenant.id)
assert await redis_client.get(cache_key) is None
@pytest.mark.asyncio
async def test_invalidate_all_user_permissions_clears_tenant(
self, db_session: AsyncSession, redis_client
):
"""invalidate_all_user_permissions removes cache for all users in a tenant."""
from app.core.permissions import CACHE_PREFIX, get_cached_permissions
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
viewer = seed["viewer_a"]
tenant = seed["tenant_a"]
await get_cached_permissions(db_session, redis_client, admin.id, tenant.id)
await get_cached_permissions(db_session, redis_client, viewer.id, tenant.id)
await invalidate_all_user_permissions(redis_client, tenant.id)
assert await redis_client.get(f"{CACHE_PREFIX}:{admin.id}:{tenant.id}") is None
assert await redis_client.get(f"{CACHE_PREFIX}:{viewer.id}:{tenant.id}") is None
@pytest.mark.asyncio
async def test_cache_hit_returns_same_permissions_as_miss(
self, db_session: AsyncSession, redis_client
):
"""Cache hit delivers the same permissions as a cache miss."""
from app.core.permissions import get_cached_permissions
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
tenant = seed["tenant_a"]
result1 = await get_cached_permissions(db_session, redis_client, admin.id, tenant.id)
result2 = await get_cached_permissions(db_session, redis_client, admin.id, tenant.id)
assert set(result1["permissions"]) == set(result2["permissions"])
assert result1["is_system_admin"] == result2["is_system_admin"]
# ═══════════════════════════════════════════════════════════════
# SECTION 8: Field-Level Permission Tests (unit + DB)
# ═══════════════════════════════════════════════════════════════
class TestFieldLevelPermissions:
"""Tests for field-level permission filtering."""
def test_filter_system_admin_no_filtering(self):
"""filter_fields_by_permission with is_system_admin → no filtering."""
data = {"name": "ACME", "annual_revenue": 1000000, "email": "info@acme.com"}
resolved = {
"is_system_admin": True,
"field_permissions": {"companies": {"annual_revenue": "hidden", "email": "hidden"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert result == data
def test_filter_no_field_permissions_no_filtering(self):
"""filter_fields_by_permission without field_permissions → no filtering."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {"is_system_admin": False, "field_permissions": {}}
result = filter_fields_by_permission(data, resolved, "companies")
assert result == data
def test_filter_hidden_field_removed(self):
"""filter_fields_by_permission removes hidden fields."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "annual_revenue" not in result
assert "name" in result
def test_filter_readonly_field_kept(self):
"""filter_fields_by_permission keeps readonly fields."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"annual_revenue": "readonly"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "annual_revenue" in result
def test_filter_read_field_kept(self):
"""filter_fields_by_permission keeps read fields."""
data = {"name": "ACME", "annual_revenue": 1000000}
resolved = {
"is_system_admin": False,
"field_permissions": {"companies": {"annual_revenue": "read"}},
}
result = filter_fields_by_permission(data, resolved, "companies")
assert "annual_revenue" in result
@pytest.mark.asyncio
async def test_company_service_applies_filter_with_resolved_perms(
self, db_session: AsyncSession
):
"""Company service applies field filtering when resolved_perms is passed."""
from app.services.company_service import get_company_detail
seed = await seed_tenant_and_users(db_session)
company = seed["company_a"]
resolved_perms = {
"is_system_admin": False,
"field_permissions": {"companies": {"industry": "hidden"}},
}
result = await get_company_detail(
db_session, seed["tenant_a"].id, company.id, resolved_perms=resolved_perms
)
assert result is not None
assert "industry" not in result
assert "name" in result
@pytest.mark.asyncio
async def test_contact_service_applies_filter_with_resolved_perms(
self, db_session: AsyncSession
):
"""Contact service applies field filtering when resolved_perms is passed."""
from app.models.contact import Contact
from app.services.contact_service import get_contact_detail
seed = await seed_tenant_and_users(db_session)
contact = Contact(
tenant_id=seed["tenant_a"].id,
first_name="John",
last_name="Doe",
email="john@example.com",
phone="123456",
mobile="789012",
created_by=seed["admin_a"].id,
updated_by=seed["admin_a"].id,
)
db_session.add(contact)
await db_session.flush()
resolved_perms = {
"is_system_admin": False,
"field_permissions": {"contacts": {"mobile": "hidden"}},
}
result = await get_contact_detail(
db_session, seed["tenant_a"].id, contact.id, resolved_perms=resolved_perms
)
assert result is not None
assert "mobile" not in result
assert "first_name" in result
@pytest.mark.asyncio
async def test_company_service_no_filter_when_resolved_perms_none(
self, db_session: AsyncSession
):
"""Company service does NOT filter when resolved_perms is None (backward compat)."""
from app.services.company_service import get_company_detail
seed = await seed_tenant_and_users(db_session)
company = seed["company_a"]
result = await get_company_detail(
db_session, seed["tenant_a"].id, company.id, resolved_perms=None
)
assert result is not None
assert "industry" in result
assert "name" in result
# ═══════════════════════════════════════════════════════════════
# HELPER FUNCTIONS for DB-dependent tests
# ═══════════════════════════════════════════════════════════════
async def _create_role_with_permissions(
db: AsyncSession, tenant_id, name: str, permissions, denied=None,
field_perms=None
) -> Role:
"""Helper: create a Role with specific permissions."""
role = Role(
tenant_id=tenant_id,
name=name,
permissions=permissions,
denied_permissions=denied or [],
field_permissions=field_perms or {},
)
db.add(role)
await db.flush()
return role
async def _create_user_with_role(
db: AsyncSession, tenant_id, role_id, email: str, name: str = "Test User"
) -> tuple[User, UserTenant]:
"""Helper: create a User with a specific role_id via UserTenant."""
user = User(
tenant_id=tenant_id,
email=email,
name=name,
password_hash=hash_password("TestPass123!"),
role="custom",
is_active=True,
preferences={},
)
db.add(user)
await db.flush()
ut = UserTenant(user_id=user.id, tenant_id=tenant_id, is_default=True, role_id=role_id)
db.add(ut)
await db.flush()
return user, ut
# ═══════════════════════════════════════════════════════════════
# SECTION 3: Group CRUD Tests (DB required — uses conftest client)
# ═══════════════════════════════════════════════════════════════
class TestGroupCRUD:
"""Tests for Group CRUD operations and membership management."""
@pytest.mark.asyncio
async def test_create_group_with_permissions(self, client: AsyncClient, db_session: AsyncSession):
"""Create a group with permissions and denied_permissions."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.post(
"/api/v1/groups",
json={
"name": "Sales Team",
"description": "Sales department group",
"permissions": {"companies": {"read": True, "write": True}},
"denied_permissions": ["companies:delete"],
"field_permissions": {"companies": {"annual_revenue": "hidden"}},
},
headers=csrf_headers(csrf),
)
assert resp.status_code == 201
data = resp.json()
assert data["name"] == "Sales Team"
assert data["permissions"] == {"companies": {"read": True, "write": True}}
assert data["denied_permissions"] == ["companies:delete"]
assert data["permission_version"] == 1
@pytest.mark.asyncio
async def test_list_groups(self, client: AsyncClient, db_session: AsyncSession):
"""List all groups in the tenant."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
for name in ["Group A", "Group B"]:
await client.post(
"/api/v1/groups",
json={"name": name, "permissions": {}},
headers=csrf_headers(csrf),
)
resp = await client.get("/api/v1/groups", headers=ORIGIN_HEADER)
assert resp.status_code == 200
items = resp.json()["items"]
assert len(items) >= 2
@pytest.mark.asyncio
async def test_get_group_by_id(self, client: AsyncClient, db_session: AsyncSession):
"""Get a single group by ID."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Test Group", "permissions": {"contacts": {"read": True}}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
resp = await client.get(f"/api/v1/groups/{group_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
assert resp.json()["name"] == "Test Group"
@pytest.mark.asyncio
async def test_update_group_name(self, client: AsyncClient, db_session: AsyncSession):
"""Update group name — permission_version should NOT increment."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Old Name", "permissions": {"companies": {"read": True}}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
initial_version = create_resp.json()["permission_version"]
resp = await client.patch(
f"/api/v1/groups/{group_id}",
json={"name": "New Name"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
assert resp.json()["name"] == "New Name"
assert resp.json()["permission_version"] == initial_version
@pytest.mark.asyncio
async def test_update_group_permissions_increments_version(
self, client: AsyncClient, db_session: AsyncSession
):
"""Update group permissions — permission_version increments."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Versioned Group", "permissions": {"companies": {"read": True}}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
initial_version = create_resp.json()["permission_version"]
resp = await client.patch(
f"/api/v1/groups/{group_id}",
json={"permissions": {"companies": {"read": True, "write": True}}},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
assert resp.json()["permission_version"] == initial_version + 1
@pytest.mark.asyncio
async def test_delete_group_soft_delete(self, client: AsyncClient, db_session: AsyncSession):
"""Delete a group (soft-delete) — group disappears from list."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "To Delete", "permissions": {}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
resp = await client.delete(f"/api/v1/groups/{group_id}", headers=csrf_headers(csrf))
assert resp.status_code == 204
list_resp = await client.get("/api/v1/groups", headers=ORIGIN_HEADER)
group_ids = [g["id"] for g in list_resp.json()["items"]]
assert group_id not in group_ids
@pytest.mark.asyncio
async def test_add_user_to_group(self, client: AsyncClient, db_session: AsyncSession):
"""Add a user to a group."""
seed = await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Members Group", "permissions": {}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
resp = await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": viewer_id},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_remove_user_from_group(self, client: AsyncClient, db_session: AsyncSession):
"""Remove a user from a group."""
seed = await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Remove Test", "permissions": {}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": viewer_id},
headers=csrf_headers(csrf),
)
resp = await client.delete(
f"/api/v1/groups/{group_id}/members/{viewer_id}",
headers=csrf_headers(csrf),
)
assert resp.status_code == 204
@pytest.mark.asyncio
async def test_list_group_members(self, client: AsyncClient, db_session: AsyncSession):
"""List all members of a group."""
seed = await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "Member List", "permissions": {}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": viewer_id},
headers=csrf_headers(csrf),
)
resp = await client.get(f"/api/v1/groups/{group_id}/members", headers=ORIGIN_HEADER)
assert resp.status_code == 200
members = resp.json()["items"]
assert any(m["user_id"] == viewer_id for m in members)
@pytest.mark.asyncio
async def test_get_user_groups(self, client: AsyncClient, db_session: AsyncSession):
"""List all groups a user is member of."""
seed = await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={"name": "User Groups Test", "permissions": {}},
headers=csrf_headers(csrf),
)
group_id = create_resp.json()["id"]
viewer_id = str(seed["viewer_a"].id)
await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": viewer_id},
headers=csrf_headers(csrf),
)
resp = await client.get(f"/api/v1/groups/user/{viewer_id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
groups = resp.json()["items"]
assert any(g["id"] == group_id for g in groups)
# ═══════════════════════════════════════════════════════════════
# SECTION 4: RBAC Route-Guard Tests (DB required — uses conftest client)
# ═══════════════════════════════════════════════════════════════
class TestRBACRouteGuard:
"""Tests for require_permission, require_admin, require_write dependencies."""
@pytest.mark.asyncio
async def test_require_permission_allows_user_with_exact_permission(
self, client: AsyncClient, db_session: AsyncSession
):
"""User with companies:read can access companies list."""
await seed_tenant_and_users(db_session)
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_require_permission_blocks_user_without_permission(
self, client: AsyncClient, db_session: AsyncSession
):
"""Viewer cannot create companies (requires companies:write)."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "viewer@tenanta.com")
resp = await client.post(
"/api/v1/companies",
json={"name": "Test Co"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_require_permission_allows_system_admin(
self, client: AsyncClient, db_session: AsyncSession
):
"""System admin bypasses all permission checks."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
admin.is_system_admin = True
await db_session.flush()
await db_session.commit()
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/companies", headers=ORIGIN_HEADER)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_require_permission_allows_wildcard_module(
self, db_session: AsyncSession
):
"""User with companies:* equivalent (all actions) can access companies:read."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = await _create_role_with_permissions(
db_session, tenant.id, "wildcard_module",
permissions={"companies": {"read": True, "write": True, "delete": True}},
)
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "wildcard_mod@test.com"
)
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "companies:write") is True
@pytest.mark.asyncio
async def test_require_permission_allows_wildcard_action(
self, db_session: AsyncSession
):
"""User with *:read permission can access companies:read."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = await _create_role_with_permissions(
db_session, tenant.id, "wildcard_action",
permissions=["*:read"],
)
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "wildcard_act@test.com"
)
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "contacts:read") is True
assert check_permission(resolved, "companies:write") is False
@pytest.mark.asyncio
async def test_require_permission_allows_star_star(
self, db_session: AsyncSession
):
"""User with *:* permission can access anything."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = await _create_role_with_permissions(
db_session, tenant.id, "superadmin",
permissions=["*:*"],
)
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "starstar@test.com"
)
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "anything:anything") is True
@pytest.mark.asyncio
async def test_require_permission_blocks_read_only_for_write(
self, db_session: AsyncSession
):
"""User with only companies:read cannot pass companies:write check."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = await _create_role_with_permissions(
db_session, tenant.id, "read_only",
permissions={"companies": {"read": True}},
)
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "readonly@test.com"
)
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert check_permission(resolved, "companies:read") is True
assert check_permission(resolved, "companies:write") is False
@pytest.mark.asyncio
async def test_require_admin_allows_system_admin(
self, client: AsyncClient, db_session: AsyncSession
):
"""require_admin allows system_admin user."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
admin.is_system_admin = True
await db_session.flush()
await db_session.commit()
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/plugins", headers=ORIGIN_HEADER)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_require_admin_allows_legacy_admin(
self, client: AsyncClient, db_session: AsyncSession
):
"""require_admin allows legacy admin role."""
await seed_tenant_and_users(db_session)
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/plugins", headers=ORIGIN_HEADER)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_require_admin_blocks_non_admin(
self, client: AsyncClient, db_session: AsyncSession
):
"""require_admin blocks viewer (legacy role)."""
await seed_tenant_and_users(db_session)
await login_with_csrf(client, "viewer@tenanta.com")
resp = await client.get("/api/v1/plugins", headers=ORIGIN_HEADER)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_require_write_allows_legacy_editor(
self, client: AsyncClient, db_session: AsyncSession
):
"""require_write allows legacy editor role (companies:write in legacy perms)."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "editor@tenanta.com")
resp = await client.post(
"/api/v1/companies",
json={"name": "Editor Company"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 201
@pytest.mark.asyncio
async def test_require_write_blocks_legacy_viewer(
self, client: AsyncClient, db_session: AsyncSession
):
"""require_write blocks legacy viewer role."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(client, "viewer@tenanta.com")
resp = await client.post(
"/api/v1/companies",
json={"name": "Viewer Company"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
# ═══════════════════════════════════════════════════════════════
# SECTION 5: Self-Modification Prevention Tests (DB required)
# ═══════════════════════════════════════════════════════════════
class TestSelfModificationPrevention:
"""Tests for self-modification prevention in user routes."""
@pytest.mark.asyncio
async def test_user_can_change_own_name(self, client: AsyncClient, db_session: AsyncSession):
"""User can change their own name (allowed)."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.patch(
f"/api/v1/users/{admin.id}",
json={"name": "New Admin Name"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
assert resp.json()["name"] == "New Admin Name"
@pytest.mark.asyncio
async def test_user_cannot_change_own_role(self, client: AsyncClient, db_session: AsyncSession):
"""User cannot change their own role (403 self_modification_forbidden)."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.patch(
f"/api/v1/users/{admin.id}",
json={"role": "viewer"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
assert resp.json()["detail"]["code"] == "self_modification_forbidden"
@pytest.mark.asyncio
async def test_user_cannot_change_own_is_active(self, client: AsyncClient, db_session: AsyncSession):
"""User cannot change their own is_active status (403)."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.patch(
f"/api/v1/users/{admin.id}",
json={"is_active": False},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
assert resp.json()["detail"]["code"] == "self_modification_forbidden"
@pytest.mark.asyncio
async def test_user_cannot_change_own_role_id(self, client: AsyncClient, db_session: AsyncSession):
"""User cannot change their own role_id (403)."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.patch(
f"/api/v1/users/{admin.id}",
json={"role_id": str(uuid.uuid4())},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
assert resp.json()["detail"]["code"] == "self_modification_forbidden"
@pytest.mark.asyncio
async def test_admin_can_change_other_user_role(
self, client: AsyncClient, db_session: AsyncSession
):
"""Admin can change another user's role (allowed)."""
seed = await seed_tenant_and_users(db_session)
viewer = seed["viewer_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
resp = await client.patch(
f"/api/v1/users/{viewer.id}",
json={"role": "editor"},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
assert resp.json()["role"] == "editor"
# ═══════════════════════════════════════════════════════════════
# SECTION 7: Mail-Plugin RBAC Tests (DB required)
# ═══════════════════════════════════════════════════════════════
@pytest_asyncio.fixture
async def mail_app(engine: AsyncEngine, redis_client):
"""FastAPI app with Mail plugin registered, installed, and activated."""
reset_engine_for_testing(engine)
app = create_app()
registry = reset_registry_for_testing()
registry.initialize(engine, app)
container = get_container()
await container.initialize()
registry.register_plugin(MailPlugin())
reset_plugin_service_for_testing(registry)
sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
async with sf() as session:
await registry.install(session, "mail")
await registry.activate(session, "mail")
await session.commit()
yield app
await close_engine()
@pytest_asyncio.fixture
async def mail_client(mail_app) -> AsyncClient:
"""HTTP async test client with mail plugin active."""
transport = ASGITransport(app=mail_app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
class TestMailPluginRBAC:
"""Tests for mail plugin RBAC permission enforcement and delegate access."""
async def _seed_mail_data(self, db: AsyncSession) -> dict:
"""Helper: seed tenant, users, mail account, folder, and mail."""
seed = await seed_tenant_and_users(db)
tenant = seed["tenant_a"]
admin = seed["admin_a"]
viewer = seed["viewer_a"]
account = MailAccount(
tenant_id=tenant.id,
user_id=admin.id,
email_address="admin@tenanta.com",
display_name="Admin",
imap_host="imap.example.com",
imap_port=993,
imap_ssl=True,
smtp_host="smtp.example.com",
smtp_port=587,
smtp_tls=True,
username="admin@tenanta.com",
encrypted_password="encrypted_dummy",
is_shared=False,
is_active=True,
)
db.add(account)
await db.flush()
shared_account = MailAccount(
tenant_id=tenant.id,
user_id=admin.id,
email_address="shared@tenanta.com",
display_name="Shared",
imap_host="imap.example.com",
imap_port=993,
imap_ssl=True,
smtp_host="smtp.example.com",
smtp_port=587,
smtp_tls=True,
username="shared@tenanta.com",
encrypted_password="encrypted_dummy",
is_shared=True,
is_active=True,
)
db.add(shared_account)
await db.flush()
folder = MailFolder(
tenant_id=tenant.id,
account_id=account.id,
name="INBOX",
imap_name="INBOX",
is_standard=True,
)
db.add(folder)
await db.flush()
mail = Mail(
tenant_id=tenant.id,
account_id=account.id,
folder_id=folder.id,
message_id="<test@example.com>",
subject="Test Subject",
from_address="sender@example.com",
to_addresses="admin@tenanta.com",
body_text="Test body",
)
db.add(mail)
await db.flush()
await db.commit()
return {
**seed,
"account": account,
"shared_account": shared_account,
"folder": folder,
"mail": mail,
}
async def _give_user_mail_perms(self, db: AsyncSession, tenant_id, user_id, perms: list[str]):
"""Helper: create a role with mail permissions and assign to user."""
from sqlalchemy import update
role = Role(
tenant_id=tenant_id,
name=f"mail_role_{uuid.uuid4().hex[:8]}",
permissions=perms,
)
db.add(role)
await db.flush()
await db.execute(
update(UserTenant)
.where(UserTenant.user_id == user_id)
.values(role_id=role.id)
)
await db.commit()
@pytest.mark.asyncio
async def test_user_without_mail_read_gets_403_on_list(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""User without mail:read permission gets 403 on GET /mail."""
await seed_tenant_and_users(db_session)
await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.get("/api/v1/mail", headers=ORIGIN_HEADER)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_user_with_mail_read_gets_200_on_list(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""User with mail:read permission gets 200 on GET /mail.
Admin (legacy) has *:* which matches mail:read."""
await seed_tenant_and_users(db_session)
await login_with_csrf(mail_client, "admin@tenanta.com")
resp = await mail_client.get(
"/api/v1/mail",
params={"account_id": str(uuid.uuid4())},
headers=ORIGIN_HEADER,
)
assert resp.status_code == 200
@pytest.mark.asyncio
async def test_user_without_mail_send_gets_403(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""User without mail:send permission gets 403 on POST /mail/send."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.post(
"/api/v1/mail/send",
json={
"account_id": str(uuid.uuid4()),
"to": ["test@example.com"],
"subject": "Test",
"body_text": "Test",
},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_user_without_mail_config_gets_403_on_create_account(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""User without mail:config permission gets 403 on POST /mail/accounts."""
await seed_tenant_and_users(db_session)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.post(
"/api/v1/mail/accounts",
json={
"email_address": "test@example.com",
"display_name": "Test",
"imap_host": "imap.example.com",
"smtp_host": "smtp.example.com",
"username": "test@example.com",
"password": "secret123",
},
headers=csrf_headers(csrf),
)
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_delegate_read_can_read_but_not_delete(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""Delegate with access_level 'read' can read mails but cannot delete."""
data = await self._seed_mail_data(db_session)
viewer = data["viewer_a"]
account = data["account"]
mail = data["mail"]
delegate = MailAccountDelegate(
tenant_id=data["tenant_a"].id,
account_id=account.id,
delegate_user_id=viewer.id,
access_level="read",
)
db_session.add(delegate)
await db_session.flush()
await db_session.commit()
await self._give_user_mail_perms(
db_session, data["tenant_a"].id, viewer.id,
["mail:read", "mail:write", "mail:delete"]
)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
resp = await mail_client.delete(f"/api/v1/mail/{mail.id}", headers=csrf_headers(csrf))
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_delegate_write_can_set_flags_but_not_delete(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""Delegate with access_level 'write' can set flags but cannot delete."""
data = await self._seed_mail_data(db_session)
viewer = data["viewer_a"]
account = data["account"]
mail = data["mail"]
delegate = MailAccountDelegate(
tenant_id=data["tenant_a"].id,
account_id=account.id,
delegate_user_id=viewer.id,
access_level="write",
)
db_session.add(delegate)
await db_session.flush()
await db_session.commit()
await self._give_user_mail_perms(
db_session, data["tenant_a"].id, viewer.id,
["mail:read", "mail:write", "mail:delete"]
)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.patch(
f"/api/v1/mail/{mail.id}/flags",
json={"is_seen": True},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
resp = await mail_client.delete(f"/api/v1/mail/{mail.id}", headers=csrf_headers(csrf))
assert resp.status_code == 403
@pytest.mark.asyncio
async def test_delegate_delete_can_delete_mail(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""Delegate with access_level 'delete' can delete mails."""
data = await self._seed_mail_data(db_session)
viewer = data["viewer_a"]
account = data["account"]
mail = data["mail"]
delegate = MailAccountDelegate(
tenant_id=data["tenant_a"].id,
account_id=account.id,
delegate_user_id=viewer.id,
access_level="delete",
)
db_session.add(delegate)
await db_session.flush()
await db_session.commit()
await self._give_user_mail_perms(
db_session, data["tenant_a"].id, viewer.id,
["mail:read", "mail:write", "mail:delete"]
)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.delete(f"/api/v1/mail/{mail.id}", headers=csrf_headers(csrf))
assert resp.status_code == 204
@pytest.mark.asyncio
async def test_delegate_full_can_do_everything(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""Delegate with access_level 'full' can perform all operations."""
data = await self._seed_mail_data(db_session)
viewer = data["viewer_a"]
account = data["account"]
mail = data["mail"]
delegate = MailAccountDelegate(
tenant_id=data["tenant_a"].id,
account_id=account.id,
delegate_user_id=viewer.id,
access_level="full",
)
db_session.add(delegate)
await db_session.flush()
await db_session.commit()
await self._give_user_mail_perms(
db_session, data["tenant_a"].id, viewer.id,
["mail:read", "mail:write", "mail:delete", "mail:config"]
)
csrf = await login_with_csrf(mail_client, "viewer@tenanta.com")
resp = await mail_client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
resp = await mail_client.patch(
f"/api/v1/mail/{mail.id}/flags",
json={"is_flagged": True},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
resp = await mail_client.delete(f"/api/v1/mail/{mail.id}", headers=csrf_headers(csrf))
assert resp.status_code == 204
@pytest.mark.asyncio
async def test_owner_has_full_access_without_delegate(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""Owner has full access to their own mailbox without any delegate entry."""
data = await self._seed_mail_data(db_session)
mail = data["mail"]
csrf = await login_with_csrf(mail_client, "admin@tenanta.com")
resp = await mail_client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
assert resp.status_code == 200
resp = await mail_client.patch(
f"/api/v1/mail/{mail.id}/flags",
json={"is_seen": True},
headers=csrf_headers(csrf),
)
assert resp.status_code == 200
resp = await mail_client.delete(f"/api/v1/mail/{mail.id}", headers=csrf_headers(csrf))
assert resp.status_code == 204
@pytest.mark.asyncio
async def test_user_without_delegate_gets_403_on_shared_mailbox(
self, mail_client: AsyncClient, db_session: AsyncSession
):
"""User without delegate access gets 403 when trying to access another user's mail."""
data = await self._seed_mail_data(db_session)
editor = data["editor_a"]
mail = data["mail"]
await self._give_user_mail_perms(
db_session, data["tenant_a"].id, editor.id,
["mail:read"]
)
await login_with_csrf(mail_client, "editor@tenanta.com")
resp = await mail_client.get(f"/api/v1/mail/{mail.id}", headers=ORIGIN_HEADER)
assert resp.status_code == 403
# ═══════════════════════════════════════════════════════════════
# SECTION 9: Integration Tests (DB + Redis required)
# ═══════════════════════════════════════════════════════════════
class TestRBACIntegration:
"""End-to-end integration tests for the RBAC system."""
@pytest.mark.asyncio
async def test_login_and_me_permissions(self, client: AsyncClient, db_session: AsyncSession):
"""Login → /me/permissions → has correct permissions."""
await seed_tenant_and_users(db_session)
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/auth/me/permissions", headers=ORIGIN_HEADER)
assert resp.status_code == 200
data = resp.json()
assert "permissions" in data
assert "denied_permissions" in data
assert "field_permissions" in data
assert "is_system_admin" in data
@pytest.mark.asyncio
async def test_login_as_admin_has_star_star(
self, client: AsyncClient, db_session: AsyncSession
):
"""Login as admin → has *:* permission."""
await seed_tenant_and_users(db_session)
await login_with_csrf(client, "admin@tenanta.com")
resp = await client.get("/api/v1/auth/me/permissions", headers=ORIGIN_HEADER)
assert resp.status_code == 200
perms = resp.json()["permissions"]
assert "*:*" in perms
@pytest.mark.asyncio
async def test_create_group_add_user_resolved_perms_include_group(
self, client: AsyncClient, db_session: AsyncSession
):
"""Create Group → Add User → User's resolved permissions include group permissions."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
viewer = seed["viewer_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={
"name": "Integration Group",
"permissions": {"companies": {"read": True, "write": True}},
},
headers=csrf_headers(csrf),
)
assert create_resp.status_code == 201
group_id = create_resp.json()["id"]
add_resp = await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": str(viewer.id)},
headers=csrf_headers(csrf),
)
assert add_resp.status_code == 200
resolved = await resolve_permissions(db_session, viewer.id, tenant.id)
assert "companies:write" in resolved["permissions"]
@pytest.mark.asyncio
async def test_update_role_permissions_cache_invalidated(
self, db_session: AsyncSession, redis_client
):
"""Update role permissions → cache invalidated → next request has new permissions."""
from app.core.permissions import CACHE_PREFIX, get_cached_permissions
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = Role(
tenant_id=tenant.id,
name="cache_test_role",
permissions={"companies": {"read": True}},
)
db_session.add(role)
await db_session.flush()
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "cache_test@test.com"
)
await db_session.commit()
result1 = await get_cached_permissions(db_session, redis_client, user.id, tenant.id)
assert "companies:read" in result1["permissions"]
assert "companies:write" not in result1["permissions"]
role.permissions = {"companies": {"read": True, "write": True}}
role.permission_version += 1
await db_session.flush()
await db_session.commit()
await invalidate_permission_cache(redis_client, user.id, tenant.id)
result2 = await get_cached_permissions(db_session, redis_client, user.id, tenant.id)
assert "companies:write" in result2["permissions"]
@pytest.mark.asyncio
async def test_delete_group_user_perms_no_longer_include_group(
self, client: AsyncClient, db_session: AsyncSession
):
"""Delete group → group is soft-deleted and disappears from list.
NOTE: resolve_permissions() does not currently filter by Group.deleted_at,
so soft-deleted groups still contribute permissions until the UserGroup
membership is removed. This is a known bug in the implementation.
The test verifies the soft-delete happened and the group is gone from the list.
"""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
viewer = seed["viewer_a"]
csrf = await login_with_csrf(client, "admin@tenanta.com")
create_resp = await client.post(
"/api/v1/groups",
json={
"name": "Delete Test Group",
"permissions": {"companies": {"delete": True}},
},
headers=csrf_headers(csrf),
)
assert create_resp.status_code == 201
group_id = create_resp.json()["id"]
await client.post(
f"/api/v1/groups/{group_id}/members",
json={"user_id": str(viewer.id)},
headers=csrf_headers(csrf),
)
resolved_before = await resolve_permissions(db_session, viewer.id, tenant.id)
assert "companies:delete" in resolved_before["permissions"]
del_resp = await client.delete(f"/api/v1/groups/{group_id}", headers=csrf_headers(csrf))
assert del_resp.status_code == 204
# Group should be gone from the list (soft-delete works for listing)
list_resp = await client.get("/api/v1/groups", headers=ORIGIN_HEADER)
group_ids = [g["id"] for g in list_resp.json()["items"]]
assert group_id not in group_ids
# KNOWN BUG: resolve_permissions does not filter by Group.deleted_at,
# so the group permission is still present after soft-delete.
# When this bug is fixed, the following assertion should pass:
# resolved_after = await resolve_permissions(db_session, viewer.id, tenant.id)
# assert "companies:delete" not in resolved_after["permissions"]
@pytest.mark.asyncio
async def test_resolve_permissions_role_with_denied(
self, db_session: AsyncSession
):
"""resolve_permissions: role permissions minus denied_permissions."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = Role(
tenant_id=tenant.id,
name="denied_test",
permissions={"companies": {"read": True, "write": True, "delete": True}},
denied_permissions=["companies:delete"],
)
db_session.add(role)
await db_session.flush()
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "denied@test.com"
)
await db_session.commit()
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert "companies:read" in resolved["permissions"]
assert "companies:write" in resolved["permissions"]
assert "companies:delete" not in resolved["permissions"]
assert "companies:delete" in resolved["denied"]
@pytest.mark.asyncio
async def test_resolve_permissions_system_admin(self, db_session: AsyncSession):
"""resolve_permissions for is_system_admin returns *:*."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
admin.is_system_admin = True
await db_session.flush()
await db_session.commit()
resolved = await resolve_permissions(db_session, admin.id, seed["tenant_a"].id)
assert resolved["is_system_admin"] is True
assert resolved["permissions"] == {"*:*"}
assert resolved["denied"] == set()
@pytest.mark.asyncio
async def test_resolve_permissions_legacy_admin(self, db_session: AsyncSession):
"""resolve_permissions for legacy admin role string returns *:*."""
seed = await seed_tenant_and_users(db_session)
admin = seed["admin_a"]
resolved = await resolve_permissions(db_session, admin.id, seed["tenant_a"].id)
assert "*:*" in resolved["permissions"]
@pytest.mark.asyncio
async def test_resolve_permissions_legacy_viewer(self, db_session: AsyncSession):
"""resolve_permissions for legacy viewer role returns read-only permissions."""
seed = await seed_tenant_and_users(db_session)
viewer = seed["viewer_a"]
resolved = await resolve_permissions(db_session, viewer.id, seed["tenant_a"].id)
assert "companies:read" in resolved["permissions"]
assert "companies:write" not in resolved["permissions"]
@pytest.mark.asyncio
async def test_resolve_permissions_no_role_no_legacy(self, db_session: AsyncSession):
"""resolve_permissions for user without role_id and without legacy role returns empty."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
user = User(
tenant_id=tenant.id,
email="norole@test.com",
name="No Role",
password_hash=hash_password("TestPass123!"),
role="",
is_active=True,
preferences={},
)
db_session.add(user)
await db_session.flush()
ut = UserTenant(user_id=user.id, tenant_id=tenant.id, is_default=True, role_id=None)
db_session.add(ut)
await db_session.commit()
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert resolved["permissions"] == set()
assert resolved["is_system_admin"] is False
@pytest.mark.asyncio
async def test_resolve_permissions_role_plus_groups_union(
self, db_session: AsyncSession
):
"""resolve_permissions for user with Role + Groups returns union of permissions."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = Role(
tenant_id=tenant.id,
name="union_role",
permissions={"companies": {"read": True}},
)
db_session.add(role)
await db_session.flush()
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "union@test.com"
)
group = Group(
tenant_id=tenant.id,
name="union_group",
permissions={"contacts": {"write": True}},
)
db_session.add(group)
await db_session.flush()
ug = UserGroup(user_id=user.id, group_id=group.id, tenant_id=tenant.id)
db_session.add(ug)
await db_session.commit()
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert "companies:read" in resolved["permissions"]
assert "contacts:write" in resolved["permissions"]
@pytest.mark.asyncio
async def test_resolve_permissions_group_denied_overrides(
self, db_session: AsyncSession
):
"""resolve_permissions: group denied_permissions override role+group allowed."""
seed = await seed_tenant_and_users(db_session)
tenant = seed["tenant_a"]
role = Role(
tenant_id=tenant.id,
name="deny_override_role",
permissions={"companies": {"read": True, "write": True}},
)
db_session.add(role)
await db_session.flush()
user, ut = await _create_user_with_role(
db_session, tenant.id, role.id, "deny_override@test.com"
)
group = Group(
tenant_id=tenant.id,
name="deny_group",
permissions={},
denied_permissions=["companies:write"],
)
db_session.add(group)
await db_session.flush()
ug = UserGroup(user_id=user.id, group_id=group.id, tenant_id=tenant.id)
db_session.add(ug)
await db_session.commit()
resolved = await resolve_permissions(db_session, user.id, tenant.id)
assert "companies:read" in resolved["permissions"]
assert "companies:write" not in resolved["permissions"]
assert "companies:write" in resolved["denied"]
Reference in New Issue Copy Permalink