feat(5.2): add User Preferences API with full-stack implementation
Backend: - Create app/models/user_preference.py with TenantMixin (user_id, key, value JSONB) - Create app/routes/user_preferences.py with GET/PUT/DELETE endpoints + RBAC - Add user_preferences:read/write to CORE_PERMISSIONS - Add user_preferences to legacy role permissions (admin/editor/viewer) - Register route in app/main.py and app/routes/__init__.py - Create alembic migration 0028_user_preferences - Add UserPreference model to conftest.py for test schema - Fix pre-existing conftest seed (Contact industry field removed in migration 0027) Frontend: - Create frontend/src/api/userPreferences.ts with React Query hooks - Create frontend/src/hooks/useUserPreferences.ts syncing with uiStore - Add i18n entries for de.json and en.json Tests: - 13 tests covering CRUD, tenant isolation, CSRF, unauthenticated access - All tests passing
This commit is contained in:
Agent Zero
@@ -0,0 +1,232 @@
|
||||
"""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=ORIGIN_HEADER)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
keys = [p["key"] for p in data["preferences"]]
|
||||
assert "theme" not in keys
|
||||
Reference in New Issue
Block a user