Agent Zero
abbe7a18fc
- P0: hooks.py 3-tuple fix, trigger_dispatcher Contract, contacts/plugin unregister_actions_by_owner - P0: 5 test files — check_permission mocks removed, hardcoded DB credential → env var - P1: attachment_service DmsFile via Contract helper, restore_registry/history_hooks dedup - P1: mail/plugin restore unregister, mcp_client datetime.now(UTC), saved_views/filters patterns - P1: ProtectedRoute fail-closed, 13 test assertion fixes (bcrypt, DB-URLs, SECRET_KEYs) - P2: deprecated notifications → post_system_message (3 files), forgejo Base, report_generator lazy import - P2: webhooks permissions, deps.py/roles.py plugin perms removed, import_export default - P2: address/tags/entity_links patterns removed, worker.py Contract-Umgehungen fixed - P2: 28 frontend TODOs (hardcoded constants, deprecated notification API) - P3: dead code, duplicates, deprecated imports, private attr, __import__ inline - P3: 8 frontend TODOs (LucideIcons, inline styles, XSS, i18n) - ruff: 838 → 0 (612 auto-fix + 246 manual + 27 F821 regression fix) - F821: 30 → 0 (AutomationDefinition, DmsFile, user_id, Path, Any, String) - Contract-Umgehungen: 2 neue gefunden (worker.py:169, worker.py:280) und gefixt
426 lines
16 KiB
Python
426 lines
16 KiB
Python
"""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."""
|
|
from app.core.auth import hash_password
|
|
|
|
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=hash_password("TestPass123!"),
|
|
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."""
|
|
from app.core.auth import hash_password
|
|
|
|
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=hash_password("TestPass123!"),
|
|
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
|
|
mail_modules = [m for m in ctx["modules"] if m["module_key"] == "mail"]
|
|
assert len(mail_modules) == 1
|
|
assert mail_modules[0]["is_visible"] == False
|
|
|
|
|
|
# ─── 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, ws_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, ws_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
|
|
|
|
|