leocrm/tests/test_workspaces.py

"""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