leocrm-bot
3ab4925783
- 10 models: tenants, users, user_tenants, roles, sessions, audit_log, deletion_log, notifications, password_reset_tokens, api_tokens - Session-based auth (Redis + PostgreSQL audit trail) - Multi-tenant with ORM-level filtering + PostgreSQL RLS (set_config) - RBAC with roles/permissions + field-level permissions - CSRF protection via Origin header validation - Auth rate limiting (Redis counters with TTL) - CORS with explicit origins (no wildcard) - Health endpoint (no auth required) - Notification service + audit log middleware - 29 tests, 26 ACs, all passing - Coverage: 62% (infrastructure modules pending coverage in later tasks)
282 lines
11 KiB
Python
282 lines
11 KiB
Python
"""Tenant + RBAC + User/Role management tests — ACs 10-17, 21-22."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from tests.conftest import ORIGIN_HEADER, seed_tenant_and_users, login_client
|
|
from app.models.audit import AuditLog
|
|
from app.models.notification import Notification
|
|
from app.models.company import Company
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserManagement:
|
|
"""ACs 10-14: User CRUD as admin/viewer."""
|
|
|
|
async def test_list_users_as_admin_returns_200(self, client: AsyncClient, db_session):
|
|
"""AC 10: GET /api/v1/users as admin -> 200 + paginated list."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.get("/api/v1/users")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert "total" in data
|
|
assert "page" in data
|
|
assert "page_size" in data
|
|
assert data["total"] >= 3 # admin, viewer, editor
|
|
|
|
async def test_list_users_as_viewer_returns_403(self, client: AsyncClient, db_session):
|
|
"""AC 11: GET /api/v1/users as viewer -> 403."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "viewer@tenanta.com")
|
|
resp = await client.get("/api/v1/users")
|
|
assert resp.status_code == 403
|
|
|
|
async def test_create_user_valid_data_returns_201(self, client: AsyncClient, db_session):
|
|
"""AC 12: POST /api/v1/users valid data -> 201."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.post(
|
|
"/api/v1/users",
|
|
json={
|
|
"email": "newuser@example.com",
|
|
"name": "New User",
|
|
"password": "NewPass123!",
|
|
"role": "viewer",
|
|
},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["email"] == "newuser@example.com"
|
|
|
|
async def test_update_user_returns_200(self, client: AsyncClient, db_session):
|
|
"""AC 13: PATCH /api/v1/users/{id} -> 200."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
# Get viewer user ID
|
|
from app.models.user import User
|
|
q = select(User).where(User.email == "viewer@tenanta.com")
|
|
result = await db_session.execute(q)
|
|
viewer = result.scalar_one()
|
|
resp = await client.patch(
|
|
f"/api/v1/users/{viewer.id}",
|
|
json={"name": "Updated Viewer"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Updated Viewer"
|
|
|
|
async def test_delete_user_returns_204(self, client: AsyncClient, db_session):
|
|
"""AC 14: DELETE /api/v1/users/{id} -> 204."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
from app.models.user import User
|
|
q = select(User).where(User.email == "viewer@tenanta.com")
|
|
result = await db_session.execute(q)
|
|
viewer = result.scalar_one()
|
|
resp = await client.delete(
|
|
f"/api/v1/users/{viewer.id}",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestRoleManagement:
|
|
"""ACs 15-16: Role list and create."""
|
|
|
|
async def test_list_roles_returns_200(self, client: AsyncClient, db_session):
|
|
"""AC 15: GET /api/v1/roles -> 200 + list with permissions."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.get("/api/v1/roles")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "items" in data
|
|
assert len(data["items"]) >= 1
|
|
assert "permissions" in data["items"][0]
|
|
assert "field_permissions" in data["items"][0]
|
|
|
|
async def test_create_role_with_custom_permissions_returns_201(self, client: AsyncClient, db_session):
|
|
"""AC 16: POST /api/v1/roles with custom permissions -> 201."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.post(
|
|
"/api/v1/roles",
|
|
json={
|
|
"name": "manager",
|
|
"permissions": {"companies": {"read": True, "create": True, "update": True, "delete": True}},
|
|
"field_permissions": {"annual_revenue": "read"},
|
|
},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
assert resp.json()["name"] == "manager"
|
|
assert "companies" in resp.json()["permissions"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestCrossTenant:
|
|
"""AC 17: Cross-tenant access on company -> 404 (not 403)."""
|
|
|
|
async def test_cross_tenant_company_access_returns_404(self, client: AsyncClient, db_session):
|
|
"""AC 17: Cross-tenant access on company -> 404 (not 403)."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
# Login as admin A (tenant A)
|
|
await login_client(client, "admin@tenanta.com")
|
|
# Try to access company B (tenant B)
|
|
company_b_id = str(seed["company_b"].id)
|
|
resp = await client.get(f"/api/v1/companies/{company_b_id}")
|
|
assert resp.status_code == 404
|
|
assert resp.status_code != 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestRBAC:
|
|
"""AC 21: RBAC — viewer can read company but not create (POST -> 403)."""
|
|
|
|
async def test_viewer_can_read_company(self, client: AsyncClient, db_session):
|
|
"""AC 21a: Viewer can GET /api/v1/companies."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "viewer@tenanta.com")
|
|
resp = await client.get("/api/v1/companies")
|
|
assert resp.status_code == 200
|
|
|
|
async def test_viewer_cannot_create_company(self, client: AsyncClient, db_session):
|
|
"""AC 21b: Viewer cannot POST /api/v1/companies -> 403."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "viewer@tenanta.com")
|
|
resp = await client.post(
|
|
"/api/v1/companies",
|
|
json={"name": "Test Company"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestFieldPermissions:
|
|
"""AC 22: Field-level permissions — hidden field not in response."""
|
|
|
|
async def test_hidden_field_not_in_response(self, client: AsyncClient, db_session):
|
|
"""AC 22: Field-level permissions — hidden field not in response.
|
|
The sales_rep role has annual_revenue as hidden.
|
|
We need a user with that role.
|
|
"""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
# Create a user with sales_rep role
|
|
from app.core.auth import hash_password
|
|
from app.models.user import User, UserTenant
|
|
sales_user = User(
|
|
tenant_id=seed["tenant_a"].id,
|
|
email="sales@tenanta.com",
|
|
name="Sales Rep",
|
|
password_hash=hash_password("TestPass123!"),
|
|
role="sales_rep",
|
|
is_active=True,
|
|
preferences={},
|
|
)
|
|
db_session.add(sales_user)
|
|
await db_session.flush()
|
|
ut = UserTenant(user_id=sales_user.id, tenant_id=seed["tenant_a"].id, is_default=True)
|
|
db_session.add(ut)
|
|
await db_session.commit()
|
|
|
|
await login_client(client, "sales@tenanta.com")
|
|
# Get company A
|
|
resp = await client.get(f"/api/v1/companies/{seed['company_a'].id}")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# annual_revenue should be filtered out for sales_rep role
|
|
assert "annual_revenue" not in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestAuditLog:
|
|
"""AC 19: Audit log entry created on company.create/update/delete."""
|
|
|
|
async def test_audit_log_on_company_create(self, client: AsyncClient, db_session):
|
|
"""AC 19: Audit log entry created on company.create."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.post(
|
|
"/api/v1/companies",
|
|
json={"name": "Audit Test Co"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
# Check audit_log table
|
|
from sqlalchemy import select as sel
|
|
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "create")
|
|
result = await db_session.execute(q)
|
|
entries = result.scalars().all()
|
|
assert len(entries) >= 1
|
|
|
|
async def test_audit_log_on_company_update(self, client: AsyncClient, db_session):
|
|
"""AC 19: Audit log entry created on company.update."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.patch(
|
|
f"/api/v1/companies/{seed['company_a'].id}",
|
|
json={"name": "Updated Company Alpha"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
from sqlalchemy import select as sel
|
|
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "update")
|
|
result = await db_session.execute(q)
|
|
entries = result.scalars().all()
|
|
assert len(entries) >= 1
|
|
|
|
async def test_audit_log_on_company_delete(self, client: AsyncClient, db_session):
|
|
"""AC 19: Audit log entry created on company.delete."""
|
|
seed = await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.delete(
|
|
f"/api/v1/companies/{seed['company_a'].id}",
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
from sqlalchemy import select as sel
|
|
q = sel(AuditLog).where(AuditLog.entity_type == "company", AuditLog.action == "delete")
|
|
result = await db_session.execute(q)
|
|
entries = result.scalars().all()
|
|
assert len(entries) >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestNotificationOnAssign:
|
|
"""AC 20: Notification created on user assign."""
|
|
|
|
async def test_notification_created_on_user_create(self, client: AsyncClient, db_session):
|
|
"""AC 20: Notification created on user assign (via user creation)."""
|
|
await seed_tenant_and_users(db_session)
|
|
await login_client(client, "admin@tenanta.com")
|
|
resp = await client.post(
|
|
"/api/v1/users",
|
|
json={
|
|
"email": "assigned@example.com",
|
|
"name": "Assigned User",
|
|
"password": "NewPass123!",
|
|
"role": "viewer",
|
|
},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
assert resp.status_code == 201
|
|
new_user_id = resp.json()["id"]
|
|
|
|
# Check notification was created for the new user
|
|
from sqlalchemy import select as sel
|
|
q = sel(Notification).where(Notification.user_id == new_user_id)
|
|
result = await db_session.execute(q)
|
|
notifs = result.scalars().all()
|
|
assert len(notifs) >= 1
|