"""Tests for Phase 6 — Workspaces. Covers: - Workspace CRUD (create, list, get, update, delete) - Module configuration (set modules, visibility) - Widget CRUD (create, list, update, delete, multiple same key) - User assignment (assign, remove, cross-tenant block) - Manager role check - Default workspace seeding - Set user default workspace - Workspace context (modules + widgets) - Empty workspace shows no modules - Cross-tenant isolation """ from __future__ import annotations import uuid import pytest from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession from app.models.tenant import Tenant from app.models.user import User, UserTenant from app.models.workspace import Workspace, WorkspaceModule, WorkspaceUser, WorkspaceWidget from app.services import workspace_service # ─── Helpers ────────────────────────────────────────────────── async def _seed_tenant_and_user(db: AsyncSession) -> dict: """Seed a tenant and a user, return IDs.""" tenant = Tenant(name="Test Tenant", slug="test-tenant") db.add(tenant) await db.flush() user = User( email="test@example.com", name="Test User", password_hash="dummy", is_active=True, preferences={}, ) db.add(user) await db.flush() ut = UserTenant( user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin", ) db.add(ut) await db.flush() return {"tenant": tenant, "user": user} async def _seed_second_tenant_and_user(db: AsyncSession) -> dict: """Seed a second tenant and user for cross-tenant tests.""" tenant = Tenant(name="Other Tenant", slug="other-tenant") db.add(tenant) await db.flush() user = User( email="other@example.com", name="Other User", password_hash="dummy", is_active=True, preferences={}, ) db.add(user) await db.flush() ut = UserTenant( user_id=user.id, tenant_id=tenant.id, is_default=True, role="admin", ) db.add(ut) await db.flush() return {"tenant": tenant, "user": user} # ─── Workspace CRUD ─────────────────────────────────────────── @pytest.mark.asyncio async def test_create_workspace(db_session: AsyncSession): """Creating a workspace returns correct data and auto-assigns creator as manager.""" seed = await _seed_tenant_and_user(db_session) result = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="Einkauf", icon="ShoppingCart", description="Einkauf-Workspace", ) assert result["name"] == "Einkauf" assert result["icon"] == "ShoppingCart" assert result["is_active"] is True assert result["user_count"] == 1 # Creator auto-assigned @pytest.mark.asyncio async def test_list_workspaces(db_session: AsyncSession): """list_workspaces returns all active workspaces for a tenant.""" seed = await _seed_tenant_and_user(db_session) await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS1", ) await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS2", ) result = await workspace_service.list_workspaces(db_session, seed["tenant"].id) assert result["total"] == 2 names = [w["name"] for w in result["items"]] assert "WS1" in names assert "WS2" in names @pytest.mark.asyncio async def test_get_workspace(db_session: AsyncSession): """get_workspace returns a single workspace with modules and user count.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) result = await workspace_service.get_workspace(db_session, seed["tenant"].id, ws_id) assert result is not None assert result["name"] == "TestWS" assert result["user_count"] == 1 @pytest.mark.asyncio async def test_update_workspace(db_session: AsyncSession): """update_workspace changes name and description.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="Original", ) ws_id = uuid.UUID(created["id"]) result = await workspace_service.update_workspace( db_session, seed["tenant"].id, ws_id, name="Updated", description="New desc", ) assert result is not None assert result["name"] == "Updated" assert result["description"] == "New desc" @pytest.mark.asyncio async def test_delete_workspace_soft_delete(db_session: AsyncSession): """delete_workspace sets is_active=False (soft delete).""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="ToDelete", ) ws_id = uuid.UUID(created["id"]) deleted = await workspace_service.delete_workspace(db_session, seed["tenant"].id, ws_id) assert deleted is True # Should not appear in list (only active workspaces) result = await workspace_service.list_workspaces(db_session, seed["tenant"].id) assert result["total"] == 0 @pytest.mark.asyncio async def test_default_workspace_uniqueness(db_session: AsyncSession): """Setting a new default workspace unsets the previous default.""" seed = await _seed_tenant_and_user(db_session) ws1 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS1", is_default=True, ) ws2 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS2", is_default=True, ) # ws1 should no longer be default result1 = await workspace_service.get_workspace(db_session, seed["tenant"].id, uuid.UUID(ws1["id"])) result2 = await workspace_service.get_workspace(db_session, seed["tenant"].id, uuid.UUID(ws2["id"])) assert result1["is_default"] is False assert result2["is_default"] is True # ─── Module Configuration ───────────────────────────────────── @pytest.mark.asyncio async def test_set_workspace_modules(db_session: AsyncSession): """set_workspace_modules replaces all modules.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) modules = [ {"module_key": "contacts", "is_visible": True, "menu_order": 10, "config": {}}, {"module_key": "calendar", "is_visible": True, "menu_order": 20, "config": {}}, {"module_key": "mail", "is_visible": False, "menu_order": 30, "config": {}}, ] result = await workspace_service.set_workspace_modules(db_session, seed["tenant"].id, ws_id, modules) assert len(result) == 3 keys = [m["module_key"] for m in result] assert "contacts" in keys assert "calendar" in keys assert "mail" in keys @pytest.mark.asyncio async def test_empty_workspace_shows_no_modules(db_session: AsyncSession): """A workspace with no modules configured shows no modules in context.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="EmptyWS", ) ws_id = uuid.UUID(created["id"]) ctx = await workspace_service.get_workspace_context( db_session, seed["tenant"].id, seed["user"].id, ws_id, ) assert ctx is not None assert ctx["modules"] == [] @pytest.mark.asyncio async def test_hidden_module_not_in_context(db_session: AsyncSession): """A module with is_visible=False does not appear in workspace context.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) modules = [ {"module_key": "contacts", "is_visible": True, "menu_order": 10, "config": {}}, {"module_key": "mail", "is_visible": False, "menu_order": 20, "config": {}}, ] await workspace_service.set_workspace_modules(db_session, seed["tenant"].id, ws_id, modules) ctx = await workspace_service.get_workspace_context( db_session, seed["tenant"].id, seed["user"].id, ws_id, ) assert ctx is not None module_keys = [m["module_key"] for m in ctx["modules"]] assert "contacts" in module_keys assert "mail" not in module_keys # Hidden module not in context # ─── Widget CRUD ────────────────────────────────────────────── @pytest.mark.asyncio async def test_create_widget(db_session: AsyncSession): """Creating a widget returns correct data.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) result = await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="recent_contacts", position_x=0, position_y=0, width=2, height=1, ) assert result["widget_key"] == "recent_contacts" assert result["width"] == 2 assert result["height"] == 1 @pytest.mark.asyncio async def test_list_widgets(db_session: AsyncSession): """list_widgets returns all widgets for a workspace.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="widget_a", position_y=0, ) await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="widget_b", position_y=1, ) result = await workspace_service.get_widgets(db_session, seed["tenant"].id, ws_id) assert len(result) == 2 @pytest.mark.asyncio async def test_multiple_same_widget_key(db_session: AsyncSession): """Multiple instances of the same widget_key can exist (no unique constraint).""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) w1 = await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="calendar_upcoming", position_x=0, position_y=0, ) w2 = await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="calendar_upcoming", position_x=1, position_y=0, ) assert w1["id"] != w2["id"] # Different IDs assert w1["widget_key"] == w2["widget_key"] # Same key result = await workspace_service.get_widgets(db_session, seed["tenant"].id, ws_id) assert len(result) == 2 # Both exist @pytest.mark.asyncio async def test_update_widget(db_session: AsyncSession): """update_widget changes position and size.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) widget = await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="test_widget", ) widget_id = uuid.UUID(widget["id"]) result = await workspace_service.update_widget( db_session, seed["tenant"].id, widget_id, position_x=2, position_y=3, width=4, height=2, ) assert result is not None assert result["position_x"] == 2 assert result["position_y"] == 3 assert result["width"] == 4 assert result["height"] == 2 @pytest.mark.asyncio async def test_delete_widget(db_session: AsyncSession): """delete_widget removes the widget.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) widget = await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="test_widget", ) widget_id = uuid.UUID(widget["id"]) deleted = await workspace_service.delete_widget(db_session, seed["tenant"].id, widget_id) assert deleted is True result = await workspace_service.get_widgets(db_session, seed["tenant"].id, ws_id) assert len(result) == 0 # ─── User Assignment ────────────────────────────────────────── @pytest.mark.asyncio async def test_assign_user_to_workspace(db_session: AsyncSession): """assign_user adds a user to a workspace.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) # Create a second user in the same tenant user2 = User( email="user2@example.com", name="User 2", password_hash="dummy", is_active=True, preferences={}, ) db_session.add(user2) await db_session.flush() ut2 = UserTenant(user_id=user2.id, tenant_id=seed["tenant"].id, is_default=True, role="viewer") db_session.add(ut2) await db_session.flush() result = await workspace_service.assign_user( db_session, seed["tenant"].id, ws_id, user2.id, role="member", ) assert result["role"] == "member" assert result["user_id"] == str(user2.id) @pytest.mark.asyncio async def test_remove_user_from_workspace(db_session: AsyncSession): """remove_user removes a user from a workspace.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) # Create a second user user2 = User( email="user2@example.com", name="User 2", password_hash="dummy", is_active=True, preferences={}, ) db_session.add(user2) await db_session.flush() ut2 = UserTenant(user_id=user2.id, tenant_id=seed["tenant"].id, is_default=True, role="viewer") db_session.add(ut2) await db_session.flush() await workspace_service.assign_user( db_session, seed["tenant"].id, ws_id, user2.id, role="member", ) removed = await workspace_service.remove_user( db_session, seed["tenant"].id, ws_id, user2.id, ) assert removed is True @pytest.mark.asyncio async def test_cross_tenant_user_assignment_blocked(db_session: AsyncSession): """verify_user_same_tenant returns False for a user from a different tenant.""" seed_a = await _seed_tenant_and_user(db_session) seed_b = await _seed_second_tenant_and_user(db_session) # User B should not be assignable to tenant A's workspace result = await workspace_service.verify_user_same_tenant( db_session, seed_a["tenant"].id, seed_b["user"].id, ) assert result is False # User A should be verified for tenant A result = await workspace_service.verify_user_same_tenant( db_session, seed_a["tenant"].id, seed_a["user"].id, ) assert result is True # ─── Manager Role Check ─────────────────────────────────────── @pytest.mark.asyncio async def test_is_workspace_manager(db_session: AsyncSession): """Creator is auto-assigned as manager; a regular member is not a manager.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) # Creator should be manager is_mgr = await workspace_service.is_workspace_manager( db_session, seed["tenant"].id, ws_id, seed["user"].id, ) assert is_mgr is True # Create a member (not manager) user2 = User( email="member@example.com", name="Member", password_hash="dummy", is_active=True, preferences={}, ) db_session.add(user2) await db_session.flush() ut2 = UserTenant(user_id=user2.id, tenant_id=seed["tenant"].id, is_default=True, role="viewer") db_session.add(ut2) await db_session.flush() await workspace_service.assign_user( db_session, seed["tenant"].id, ws_id, user2.id, role="member", ) is_mgr2 = await workspace_service.is_workspace_manager( db_session, seed["tenant"].id, ws_id, user2.id, ) assert is_mgr2 is False # ─── Default Workspace Seeding ──────────────────────────────── @pytest.mark.asyncio async def test_seed_default_workspace(db_session: AsyncSession): """seed_default_workspace creates a default workspace with all standard modules.""" seed = await _seed_tenant_and_user(db_session) result = await workspace_service.seed_default_workspace( db_session, seed["tenant"].id, seed["user"].id, ) assert result is not None assert result["name"] == "Standard" assert result["is_default"] is True assert result["user_count"] == 1 @pytest.mark.asyncio async def test_seed_default_workspace_idempotent(db_session: AsyncSession): """seed_default_workspace returns None if workspaces already exist.""" seed = await _seed_tenant_and_user(db_session) # First call creates the default workspace await workspace_service.seed_default_workspace( db_session, seed["tenant"].id, seed["user"].id, ) # Second call should return None (already has workspaces) result = await workspace_service.seed_default_workspace( db_session, seed["tenant"].id, seed["user"].id, ) assert result is None # ─── Set User Default Workspace ─────────────────────────────── @pytest.mark.asyncio async def test_set_user_default_workspace(db_session: AsyncSession): """set_user_default_workspace sets a workspace as default and unsets others.""" seed = await _seed_tenant_and_user(db_session) ws1 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS1", ) ws2 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="WS2", ) ws1_id = uuid.UUID(ws1["id"]) ws2_id = uuid.UUID(ws2["id"]) # Assign user to both # (creator is already auto-assigned to both as manager) # Set WS2 as default await workspace_service.set_user_default_workspace( db_session, seed["tenant"].id, seed["user"].id, ws2_id, ) # Verify via get_my_workspaces my = await workspace_service.get_my_workspaces( db_session, seed["tenant"].id, seed["user"].id, ) for item in my["items"]: if item["id"] == ws2["id"]: assert item["is_user_default"] is True elif item["id"] == ws1["id"]: assert item["is_user_default"] is False # ─── Workspace Context ──────────────────────────────────────── @pytest.mark.asyncio async def test_workspace_context_includes_modules_and_widgets(db_session: AsyncSession): """get_workspace_context returns both modules and widgets.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) # Set modules await workspace_service.set_workspace_modules( db_session, seed["tenant"].id, ws_id, [{"module_key": "contacts", "is_visible": True, "menu_order": 10, "config": {}}], ) # Create widget await workspace_service.create_widget( db_session, seed["tenant"].id, ws_id, widget_key="recent_contacts", ) ctx = await workspace_service.get_workspace_context( db_session, seed["tenant"].id, seed["user"].id, ws_id, ) assert ctx is not None assert len(ctx["modules"]) == 1 assert ctx["modules"][0]["module_key"] == "contacts" assert len(ctx["widgets"]) == 1 assert ctx["widgets"][0]["widget_key"] == "recent_contacts" @pytest.mark.asyncio async def test_workspace_context_unassigned_user(db_session: AsyncSession): """get_workspace_context returns None for an unassigned user.""" seed = await _seed_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="TestWS", ) ws_id = uuid.UUID(created["id"]) # Create a second user NOT assigned to the workspace user2 = User( email="unassigned@example.com", name="Unassigned", password_hash="dummy", is_active=True, preferences={}, ) db_session.add(user2) await db_session.flush() ut2 = UserTenant(user_id=user2.id, tenant_id=seed["tenant"].id, is_default=True, role="viewer") db_session.add(ut2) await db_session.flush() ctx = await workspace_service.get_workspace_context( db_session, seed["tenant"].id, user2.id, ws_id, ) assert ctx is None # User not assigned # ─── Cross-Tenant Isolation ─────────────────────────────────── @pytest.mark.asyncio async def test_cross_tenant_workspace_isolation(db_session: AsyncSession): """A workspace from tenant A cannot be accessed by tenant B.""" seed_a = await _seed_tenant_and_user(db_session) seed_b = await _seed_second_tenant_and_user(db_session) created = await workspace_service.create_workspace( db_session, seed_a["tenant"].id, seed_a["user"].id, name="TenantA_WS", ) ws_id = uuid.UUID(created["id"]) # Tenant B should not see tenant A's workspace result = await workspace_service.get_workspace( db_session, seed_b["tenant"].id, ws_id, ) assert result is None # Not found in tenant B @pytest.mark.asyncio async def test_get_my_workspaces_only_assigned(db_session: AsyncSession): """get_my_workspaces returns only workspaces the user is assigned to.""" seed = await _seed_tenant_and_user(db_session) ws1 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="AssignedWS", ) ws2 = await workspace_service.create_workspace( db_session, seed["tenant"].id, seed["user"].id, name="AlsoAssignedWS", ) # Create a third workspace without assigning the user ws3 = Workspace( tenant_id=seed["tenant"].id, name="UnassignedWS", is_active=True, created_by=seed["user"].id, ) db_session.add(ws3) await db_session.flush() my = await workspace_service.get_my_workspaces( db_session, seed["tenant"].id, seed["user"].id, ) names = [w["name"] for w in my["items"]] assert "AssignedWS" in names assert "AlsoAssignedWS" in names assert "UnassignedWS" not in names # User not assigned