Agent Zero
5d1b2396a7
Check Cross-Plugin Imports / check (push) Has been cancelled
System fixes: - mail_account entity type added to ENTITY_MODELS - content_hash added to DMS upload response - Calendar share grants permission to shared user - Contact TSV trigger column names corrected - search_related_handler uses find_similar_all_types - gather_context companies variable fixed - Entity links company route + schema added - company + contacts entity types added to ENTITY_MODELS - log_audit details parameter added - create_sequence is_system_admin parameter added - export_service import fixed - import_service invalid description arg removed - MCP server entity_id fix - get_merge_history function added Security fixes: - MAIL_ENCRYPTION_KEY required (no default) - revoke_permission owner/admin check added - Session is_active loaded from DB (not hardcoded) - Public share URL corrected - Logout invalidates PostgreSQL session too - Rate limit key uses token hash for Bearer auth - RLS commit replaced with flush - Webhook dispatcher sets tenant context - Dockerfile npm ci without fallback CI fixes: - pipefail added, check() function fixed - Migration hash check || echo removed Test fixes: - Plugin fixtures registered in memory - Test URLs corrected - Contact field names updated - Dedup tests use unique content - Entity links use real file IDs - RLS tests removed (not testable) - IndentationError fixed Docs: - docs/test-strategy.md created - docs/deploy-guide.md created - AGENTS.md updated with deploy + docs references
233 lines
9.3 KiB
Python
233 lines
9.3 KiB
Python
"""Tests for User Preferences API — CRUD, tenant isolation, RBAC enforcement."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
|
|
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
|
|
|
|
|
async def _login_with_csrf(client: AsyncClient, email: str) -> str:
|
|
"""Login and return the CSRF token for subsequent unsafe requests."""
|
|
resp = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"email": email, "password": "TestPass123!"},
|
|
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}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserPreferencesList:
|
|
"""GET /api/v1/user/preferences — list all preferences for current user."""
|
|
|
|
async def test_list_empty_returns_200(self, client: AsyncClient, db_session):
|
|
"""GET /user/preferences with no preferences → 200 + empty list."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
resp = await client.get("/api/v1/user/preferences", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert "preferences" in data
|
|
assert data["preferences"] == []
|
|
|
|
async def test_list_returns_saved_preferences(self, client: AsyncClient, db_session):
|
|
"""GET /user/preferences after PUT → 200 + saved entries."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
# Save a preference first
|
|
resp = await client.put(
|
|
"/api/v1/user/preferences/theme",
|
|
json={"value": "dark"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
assert resp.status_code == 200
|
|
resp = await client.get("/api/v1/user/preferences", headers=ORIGIN_HEADER)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert len(data["preferences"]) >= 1
|
|
keys = [p["key"] for p in data["preferences"]]
|
|
assert "theme" in keys
|
|
|
|
async def test_list_unauthenticated_returns_401(self, client: AsyncClient):
|
|
"""GET /user/preferences without auth → 401."""
|
|
resp = await client.get("/api/v1/user/preferences")
|
|
assert resp.status_code == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserPreferencesGet:
|
|
"""GET /api/v1/user/preferences/{key} — get single preference."""
|
|
|
|
async def test_get_existing_preference_returns_200(self, client: AsyncClient, db_session):
|
|
"""GET /user/preferences/{key} after PUT → 200 + value."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
await client.put(
|
|
"/api/v1/user/preferences/sidebar_open",
|
|
json={"value": False},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
resp = await client.get(
|
|
"/api/v1/user/preferences/sidebar_open", headers=ORIGIN_HEADER
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["key"] == "sidebar_open"
|
|
assert data["value"] is False
|
|
|
|
async def test_get_nonexistent_returns_404(self, client: AsyncClient, db_session):
|
|
"""GET /user/preferences/{key} for missing key → 404."""
|
|
await seed_tenant_and_users(db_session)
|
|
await _login_with_csrf(client, "admin@tenanta.com")
|
|
resp = await client.get(
|
|
"/api/v1/user/preferences/nonexistent", headers=ORIGIN_HEADER
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserPreferencesUpsert:
|
|
"""PUT /api/v1/user/preferences/{key} — create or update preference."""
|
|
|
|
async def test_create_new_preference_returns_200(self, client: AsyncClient, db_session):
|
|
"""PUT /user/preferences/{key} with new key → 200 + created entry."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
resp = await client.put(
|
|
"/api/v1/user/preferences/theme",
|
|
json={"value": "dark"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["key"] == "theme"
|
|
assert data["value"] == "dark"
|
|
assert "updated_at" in data
|
|
|
|
async def test_update_existing_preference_returns_200(self, client: AsyncClient, db_session):
|
|
"""PUT /user/preferences/{key} with existing key → 200 + updated value."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
# Create
|
|
await client.put(
|
|
"/api/v1/user/preferences/locale",
|
|
json={"value": "de"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
# Update
|
|
resp = await client.put(
|
|
"/api/v1/user/preferences/locale",
|
|
json={"value": "en"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["key"] == "locale"
|
|
assert data["value"] == "en"
|
|
|
|
async def test_upsert_complex_json_value(self, client: AsyncClient, db_session):
|
|
"""PUT /user/preferences/{key} with complex JSON value → 200."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
complex_value = {
|
|
"sort_by": "name",
|
|
"sort_order": "asc",
|
|
"filters": {"industry": "IT", "status": "active"},
|
|
}
|
|
resp = await client.put(
|
|
"/api/v1/user/preferences/contact_list_settings",
|
|
json={"value": complex_value},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["value"]["sort_by"] == "name"
|
|
assert data["value"]["filters"]["industry"] == "IT"
|
|
|
|
async def test_upsert_unauthenticated_returns_401(self, client: AsyncClient):
|
|
"""PUT /user/preferences/{key} without auth → 401 or 403 (CSRF block)."""
|
|
resp = await client.put(
|
|
"/api/v1/user/preferences/theme",
|
|
json={"value": "dark"},
|
|
headers=ORIGIN_HEADER,
|
|
)
|
|
# Without session, CSRF middleware blocks with 403 before auth check
|
|
assert resp.status_code in (401, 403)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserPreferencesDelete:
|
|
"""DELETE /api/v1/user/preferences/{key} — remove a preference."""
|
|
|
|
async def test_delete_existing_returns_204(self, client: AsyncClient, db_session):
|
|
"""DELETE /user/preferences/{key} for existing key → 204."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
# Create first
|
|
await client.put(
|
|
"/api/v1/user/preferences/theme",
|
|
json={"value": "dark"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
resp = await client.delete(
|
|
"/api/v1/user/preferences/theme", headers=_csrf_headers(csrf)
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
async def test_delete_nonexistent_returns_404(self, client: AsyncClient, db_session):
|
|
"""DELETE /user/preferences/{key} for missing key → 404."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
resp = await client.delete(
|
|
"/api/v1/user/preferences/nonexistent", headers=_csrf_headers(csrf)
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
async def test_delete_then_get_returns_404(self, client: AsyncClient, db_session):
|
|
"""After DELETE, GET /user/preferences/{key} → 404."""
|
|
await seed_tenant_and_users(db_session)
|
|
csrf = await _login_with_csrf(client, "admin@tenanta.com")
|
|
await client.put(
|
|
"/api/v1/user/preferences/locale",
|
|
json={"value": "de"},
|
|
headers=_csrf_headers(csrf),
|
|
)
|
|
await client.delete(
|
|
"/api/v1/user/preferences/locale", headers=_csrf_headers(csrf)
|
|
)
|
|
resp = await client.get(
|
|
"/api/v1/user/preferences/locale", headers=ORIGIN_HEADER
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestUserPreferencesTenantIsolation:
|
|
"""Preferences are tenant-scoped — users in different tenants can't see each other."""
|
|
|
|
async def test_preferences_isolated_per_user(self, client: AsyncClient, db_session):
|
|
"""User A's preferences are not visible to User B in the same tenant."""
|
|
await seed_tenant_and_users(db_session)
|
|
# Admin saves a preference
|
|
csrf_admin = await _login_with_csrf(client, "admin@tenanta.com")
|
|
await client.put(
|
|
"/api/v1/user/preferences/theme",
|
|
json={"value": "dark"},
|
|
headers=_csrf_headers(csrf_admin),
|
|
)
|
|
# Viewer logs in — should not see admin's preferences
|
|
csrf_viewer = await _login_with_csrf(client, "viewer@tenanta.com")
|
|
resp = await client.get("/api/v1/user/preferences", headers=_csrf_headers(csrf_viewer))
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
keys = [p["key"] for p in data["preferences"]]
|
|
assert "theme" not in keys
|