leocrm/tests/conftest.py

"""Test fixtures: PostgreSQL test DB, Redis, async test client, auth helpers.

Each test gets a fresh database schema (created from metadata) and a clean Redis.
Auth helpers talk to the HTTP API (integration tests).
"""

from __future__ import annotations

import os
import shutil
import subprocess
import sys

# Override .env settings for tests — must be set BEFORE any app imports
# so that pydantic-settings picks them up on first get_settings() call
os.environ["SESSION_COOKIE_SECURE"] = "false"
os.environ["SESSION_COOKIE_SAMESITE"] = "lax"
os.environ["SECRET_KEY"] = "test-secret-key-with-at-least-32-characters-for-testing-only!!"
os.environ["ENVIRONMENT"] = "testing"
os.environ["MIGRATION_DATABASE_URL"] = "postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm_test"
os.environ.setdefault("MAIL_ENCRYPTION_KEY", "test-mail-encryption-key")

from collections.abc import AsyncGenerator
from typing import Any

import pytest
import pytest_asyncio
import redis.asyncio as aioredis
from httpx import ASGITransport, AsyncClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import (
    AsyncEngine,
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)

from app.core.auth import hash_password
from app.core.db import Base, close_engine, reset_engine_for_testing
from app.core.service_container import get_container  # noqa: F401
from app.main import create_app

try:
    from app.models.ai_conversation import AIConversation, AIMessage  # noqa: F401
except ImportError:
    pass
from app.ai.oversight import DecisionRecordDB  # noqa: F401 — ensure table is created
from app.models.compliance import ComplianceIncident  # noqa: F401
from app.models.consumer_inbox import ConsumerInbox  # noqa: F401
from app.models.contact import Contact, ContactPerson  # noqa: F401
from app.models.contact_merge import ContactMergeHistory  # noqa: F401
from app.models.outbox import EventOutbox  # noqa: F401
from app.models.plugin import Plugin, PluginMigration  # noqa: F401
from app.models.role import Role
from app.models.saved_filter import SavedFilter  # noqa: F401
from app.models.tenant import Tenant
from app.models.user import User, UserTenant
from app.models.user_preference import UserPreference  # noqa: F401
from app.models.workflow import Workflow, WorkflowInstance, WorkflowStepHistory  # noqa: F401
from app.plugins.builtins.calendar import CalendarPlugin  # noqa: F401

# Dynamically import all plugin models so Base.metadata.create_all() includes their tables.
# This replaces ~30 hardcoded plugin imports with dynamic discovery (P1-14 fix).
from app.plugins.registry import get_registry

_registry = get_registry()
_registry.discover_builtins()
for _plugin_name in _registry.list_discovered():
    _plugin = _registry.get_plugin(_plugin_name)
    if _plugin is not None:
        # Importing get_entity_models() triggers model class imports
        # which registers them with Base.metadata
        _plugin.get_entity_models()
        # Also import the plugin's __init__ to ensure all models are loaded
        import importlib
        try:
            importlib.import_module(f"app.plugins.builtins.{_plugin_name}")
        except Exception:
            pass
        # Also directly import models.py to ensure all tables are registered
        try:
            importlib.import_module(f"app.plugins.builtins.{_plugin_name}.models")
        except Exception:
            pass

# Also import core models that may be missing
# Wiki plugin models — not loaded by get_entity_models()
from app.core.permission_registry import init_permission_registry  # noqa: F401

# Knowledge plugin models — new plugin, ensure table is created in test-DB
from app.plugins.builtins.knowledge.models import KnowledgeExtraction  # noqa: F401

# Self-improvement plugin models — new plugin, ensure tables are created in test-DB
from app.plugins.builtins.self_improvement.models import (  # noqa: F401
    ImpactMeasurement,
    ImprovementPattern,
    ImprovementProposal,
    ImprovementSignal,
)
from app.plugins.builtins.wiki.models import (  # noqa: F401
    WikiArticle,
    WikiArticleVersion,
    WikiCategory,
)
from app.plugins.registry import reset_registry_for_testing  # noqa: F401
from app.services.plugin_service import reset_plugin_service_for_testing  # noqa: F401

# Import plugin models so Base.metadata.create_all includes their tables

TEST_DB_URL = "postgresql+asyncpg://leocrm:leocrm@localhost:5432/leocrm_test"

# Clear settings cache so the env overrides (set at top of file) take effect
from app.config import get_settings

get_settings.cache_clear()


def _get_sync_engine():
    """Create a sync engine for DDL operations (drop/create schema).
    Uses postgres superuser because leocrm user doesn't own the public schema.
    """
    from sqlalchemy import create_engine

    return create_engine(
        "postgresql+psycopg2://postgres@localhost:5432/leocrm_test",
        echo=False,
    )


def _run_migrations():
    """Run alembic upgrade head to create schema from migrations.

    This replaces Base.metadata.create_all() so the test schema matches
    production (which uses Alembic migrations).  Uses subprocess to invoke
    the Alembic CLI with the test database URL.
    """
    project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
    env = os.environ.copy()
    # MIGRATION_DATABASE_URL is already set at top of file for the test DB
    result = subprocess.run(
        [sys.executable, "-m", "alembic", "upgrade", "head"],
        cwd=project_root,
        env=env,
        capture_output=True,
        text=True,
        timeout=120,
    )
    if result.returncode != 0:
        print(f"[CONFTEST] alembic upgrade head FAILED (rc={result.returncode})")
        print(f"[CONFTEST] stdout: {result.stdout}")
        print(f"[CONFTEST] stderr: {result.stderr}")
        raise RuntimeError(
            f"alembic upgrade head failed: {result.stderr or result.stdout}"
        )
    print("[CONFTEST] alembic upgrade head completed successfully")


@pytest.fixture(scope="session", autouse=True)
def db_setup():
    """Drop and recreate all tables once per test session via Alembic migrations.

    Always drops the public schema and recreates it, then runs
    ``alembic upgrade head`` so the test schema matches production exactly.
    This replaces the previous ``Base.metadata.create_all()`` approach which
    created tables from model definitions and could drift from migrations.
    """
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        # Create crm_user role if missing (needed by some migrations)
        conn.execute(text("DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_user') THEN CREATE ROLE crm_user LOGIN PASSWORD 'leocrm'; END IF; END $$;"))
        # Set a short lock timeout to prevent deadlocks
        conn.execute(text("SET lock_timeout = '10s';"))
        try:
            conn.execute(text("DROP SCHEMA IF EXISTS public CASCADE;"))
            conn.execute(text("CREATE SCHEMA public;"))
            conn.execute(text("GRANT ALL ON SCHEMA public TO leocrm;"))
            # Plugin models use the pgvector Vector type (contacts.embedding);
            # the extension lives at database level and must exist before
            # alembic creates those tables.
            conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
        except Exception:
            conn.rollback()
            conn.execute(text("SET lock_timeout = '10s';"))
            conn.execute(text(
                "DO $$ DECLARE r RECORD; BEGIN "
                "FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname='public') "
                "LOOP EXECUTE 'DROP TABLE public.' || quote_ident(r.tablename) || ' CASCADE'; "
                "END LOOP; END $$;"
            ))
        conn.commit()
    sync_eng.dispose()

    # Create all tables from models (same as sync_plugin_schema.py in production).
    # This creates ALL tables including plugin tables that have no Alembic migration.
    # In production, prestart.sh runs alembic upgrade head first, then sync_plugin_schema.py
    # runs create_all. In tests, we run create_all only because alembic migrations
    # conflict with create_all (migrations try to CREATE tables that already exist).
    # The schema drifts (VARCHAR lengths, RLS policies) are fixed by migrations 0134-0136
    # which run in production via prestart.sh.
    print("[CONFTEST] Running create_all for all model tables...")
    sync_eng2 = _get_sync_engine()
    with sync_eng2.connect() as conn:
        Base.metadata.create_all(conn, checkfirst=True)
        conn.commit()
    sync_eng2.dispose()
    print("[CONFTEST] create_all completed.")

    # Fix contacts_tsv_trigger: ensure correct column names (firstname, not first_name)
    print("[CONFTEST] Fixing contacts_tsv_trigger...")
    try:
        sync_eng2 = _get_sync_engine()
        with sync_eng2.connect() as conn:
            conn.execute(text("SET search_path TO public;"))
            conn.execute(text("DROP TRIGGER IF EXISTS contacts_tsv_update ON contacts;"))
            conn.execute(text("DROP FUNCTION IF EXISTS contacts_tsv_trigger();"))
            conn.execute(text("""
                CREATE OR REPLACE FUNCTION contacts_tsv_trigger() RETURNS trigger AS $$
                BEGIN
                    NEW.search_tsv :=
                        setweight(to_tsvector('pg_catalog.german',
                            coalesce(NEW.name, '') || ' ' || coalesce(NEW.displayname, '') ||
                            ' ' || coalesce(NEW.firstname, '') || ' ' || coalesce(NEW.surname, '')), 'A') ||
                        setweight(to_tsvector('pg_catalog.german',
                            coalesce(NEW.email_1, '') || ' ' || coalesce(NEW.email_2, '')), 'B') ||
                        setweight(to_tsvector('pg_catalog.german',
                            coalesce(NEW.phone_1, '') || ' ' || coalesce(NEW.phone_2, '')), 'C') ||
                        setweight(to_tsvector('pg_catalog.german',
                            coalesce(NEW.code, '') || ' ' || coalesce(NEW.mailing_city, '') ||
                            ' ' || coalesce(NEW.mailing_postalcode, '') || ' ' || coalesce(NEW.tags, '') ||
                            ' ' || coalesce(NEW.projectnote, '')), 'D');
                    RETURN NEW;
                END;
                $$ LANGUAGE plpgsql;
            """))
            conn.execute(text("""
                CREATE TRIGGER contacts_tsv_update
                    BEFORE INSERT OR UPDATE ON contacts
                    FOR EACH ROW EXECUTE FUNCTION contacts_tsv_trigger();
            """))
            conn.commit()
        sync_eng2.dispose()
        print("[CONFTEST] Trigger fix applied successfully")
    except Exception as e:
        print(f"[CONFTEST] Trigger fix FAILED: {e}")

    # Grant crm_api role access + enable RLS + create tenant isolation
    # policies (Block E / I-C): The Cross-Tenant Security tests connect as
    # crm_api (NOSUPERUSER, NOBYPASSRLS) to verify RLS enforcement.
    print("[CONFTEST] Setting up RLS grants and policies...")
    try:
        sync_eng3 = _get_sync_engine()
        with sync_eng3.connect() as conn:
            # 1. Ensure crm_api role exists
            conn.execute(text(
                "DO $$ BEGIN "
                "IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_api') THEN "
                "CREATE ROLE crm_api LOGIN PASSWORD 'crm_api_password' NOSUPERUSER NOBYPASSRLS; "
                "END IF; END $$;"
            ))

            # 2. Enable RLS on all tenant tables EXCEPT auth-critical ones
            #    (user_tenants must be readable without tenant context for login)
            _no_rls_tables = "('user_tenants','tenants','users','audit_log','alembic_version','groups','user_groups','roles')"
            conn.execute(text(f"""
                DO $$ DECLARE r RECORD;
                BEGIN
                    FOR r IN (
                        SELECT c.relname AS tablename
                        FROM pg_class c
                        JOIN pg_namespace n ON n.oid = c.relnamespace
                        WHERE n.nspname = 'public'
                          AND c.relkind = 'r'
                          AND c.relrowsecurity = false
                          AND c.relname NOT IN {_no_rls_tables}
                          AND EXISTS (
                              SELECT 1 FROM information_schema.columns ic
                              WHERE ic.table_schema = 'public'
                                AND ic.table_name = c.relname
                                AND ic.column_name = 'tenant_id'
                          )
                    ) LOOP
                        EXECUTE format('ALTER TABLE public.%I ENABLE ROW LEVEL SECURITY', r.tablename);
                    END LOOP;
                END $$;
            """))

            # 3. Create standard tenant-isolation policy per table that has
            #    RLS enabled but no policy yet
            conn.execute(text("""
                DO $$ DECLARE r RECORD;
                BEGIN
                    FOR r IN (
                        SELECT c.relname AS tablename
                        FROM pg_class c
                        JOIN pg_namespace n ON n.oid = c.relnamespace
                        WHERE n.nspname = 'public'
                          AND c.relkind = 'r'
                          AND c.relrowsecurity = true
                          AND NOT EXISTS (
                              SELECT 1 FROM pg_policies p
                              WHERE p.schemaname = 'public'
                                AND p.tablename = c.relname
                          )
                    ) LOOP
                        EXECUTE format(
                            'CREATE POLICY %I_tenant_isolation ON public.%I '
                            'FOR ALL USING (tenant_id = current_setting(''app.current_tenant_id'', true)::uuid) '
                            'WITH CHECK (tenant_id = current_setting(''app.current_tenant_id'', true)::uuid)',
                            r.tablename, r.tablename
                        );
                    END LOOP;
                END $$;
            """))

            # 4. Grant crm_api access to all tables
            conn.execute(text("GRANT USAGE ON SCHEMA public TO crm_api"))
            conn.execute(text("GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO crm_api"))
            conn.execute(text("GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO crm_api"))
            conn.commit()

        # Count results
        with sync_eng3.connect() as verify_conn:
            rls_count = verify_conn.execute(text(
                "SELECT count(*) FROM pg_tables WHERE schemaname='public' AND rowsecurity=true"
            )).scalar()
            pol_count = verify_conn.execute(text(
                "SELECT count(*) FROM pg_policies WHERE schemaname='public'"
            )).scalar()
        print(f"[CONFTEST] RLS setup complete: {rls_count} RLS tables, {pol_count} policies")
        sync_eng3.dispose()
    except Exception as e:
        print(f"[CONFTEST] RLS setup FAILED: {e}")


    yield

    # Cleanup after session — use TRUNCATE instead of DROP to avoid deadlocks
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        conn.execute(text("SET lock_timeout = '5s';"))
        try:
            conn.execute(text(
                "DO $$ DECLARE r RECORD; BEGIN "
                "FOR r IN (SELECT tablename FROM pg_tables WHERE schemaname='public') "
                "LOOP EXECUTE 'TRUNCATE TABLE public.' || quote_ident(r.tablename) || ' CASCADE'; "
                "END LOOP; END $$;"
            ))
        except Exception:
            pass
        conn.commit()
    sync_eng.dispose()


@pytest.fixture(autouse=True)
def clean_tables(db_setup):
    """Clean all table data before each test (preserve schema).

    Dynamically builds the TRUNCATE list from tables that actually exist
    in the database, so plugin tables that were not created (e.g. when
    only core model tables are present) do not cause errors.
    """
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        conn.execute(text("SET lock_timeout = '30s';"))
        # Query existing table names, exclude problematic tables
        result = conn.execute(
            text(
                "SELECT table_name FROM information_schema.tables "
                "WHERE table_schema = 'public' AND table_type = 'BASE TABLE' "
                "AND table_name NOT IN ('alembic_version', 'unified_search_index_log', 'unified_search_providers')"
            )
        )
        existing_tables = [row[0] for row in result]
        if existing_tables:
            table_list = ", ".join(existing_tables)
            conn.execute(text(f"TRUNCATE TABLE {table_list} CASCADE;"))
        conn.commit()
    sync_eng.dispose()
    yield


@pytest_asyncio.fixture(scope="session")
async def redis_client() -> AsyncGenerator[aioredis.Redis, None]:
    """Redis client for tests — flushes DB before and after."""
    r = aioredis.from_url("redis://localhost:6379/0", decode_responses=True)
    await r.flushdb()
    yield r
    await r.flushdb()
    await r.aclose()


# ─── Global IMAP mocking (Block I-E) ────────────────────────────────────────
#
# The mail service layer opens real aioimaplib.IMAP4_SSL connections to the
# account's imap_host (test fixtures use imap.example.com). Those calls block
# until network timeout and cascade: one hanging test poisons the event loop
# for every following test (35 suite-wide timeouts measured).
#
# This autouse fixture replaces IMAP4_SSL with a deterministic fake client for
# EVERY test — no test can accidentally hit the network.


class _FakeIMAPResponse:
    """Mimics aioimaplib response objects: tuple-like + .result attribute."""

    def __init__(self, result="OK", lines: list | None = None):
        self.result = result
        self.lines = lines or []

    def __getitem__(self, idx):
        if idx == 0:
            return self.result
        return self.lines[idx - 1] if 0 < idx <= len(self.lines) else []


def _build_fake_imap_client():
    from unittest.mock import AsyncMock, MagicMock

    client = MagicMock()
    client.wait_hello_from_server = AsyncMock(return_value=None)
    client.login = AsyncMock(return_value=_FakeIMAPResponse("OK", [b"Logged in"]))
    client.logout = AsyncMock(return_value=_FakeIMAPResponse("OK", [b"Bye"]))
    # select(folder) → OK with EXISTS count
    client.select = AsyncMock(
        return_value=_FakeIMAPResponse("OK", [b"0"])
    )
    # uid_search(...) → response whose [1][0] is a space-separated uid bytes list
    client.uid_search = AsyncMock(
        return_value=_FakeIMAPResponse("OK", [b""])
    )
    # uid('fetch', ...) → minimal RFC822 envelope; services parse defensively
    fetch_resp = _FakeIMAPResponse(
        "OK",
        [
            (
                b"1 (RFC822 {5}",
                b"Subject: t\r\n\r\nbody",
            ),
            b")",
        ],
    )
    client.uid = AsyncMock(return_value=fetch_resp)
    client.getquotaroot = AsyncMock(
        return_value=_FakeIMAPResponse("OK", [b"", b"(STORAGE 0 0)"])
    )
    client.list = AsyncMock(
        return_value=_FakeIMAPResponse("OK", [])
    )
    client.append = AsyncMock(return_value=_FakeIMAPResponse("OK", [b"Appended"]))
    return client


@pytest.fixture(autouse=True)
def mock_imap_connections(monkeypatch):
    """Replace aioimaplib.IMAP4_SSL everywhere with a deterministic fake.

    Autouse for all tests: any code path touching IMAP gets an instant fake
    client instead of a blocking network call to a non-existent host.
    """
    fake_client = _build_fake_imap_client()

    def _fake_factory(*args, **kwargs):
        return fake_client

    monkeypatch.setattr(
        "app.plugins.builtins.mail.services.aioimaplib.IMAP4_SSL",
        _fake_factory,
    )
    yield fake_client


@pytest_asyncio.fixture(scope="session")
async def engine() -> AsyncGenerator[AsyncEngine, None]:
    """Async engine for the test database (session-scoped for speed)."""
    eng = create_async_engine(TEST_DB_URL, echo=False)
    yield eng
    await eng.dispose()


@pytest_asyncio.fixture
async def session_factory(engine: AsyncEngine) -> async_sessionmaker[AsyncSession]:
    """Session factory bound to the test engine."""
    return async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)


@pytest_asyncio.fixture
async def db_session(
    session_factory: async_sessionmaker[AsyncSession],
) -> AsyncGenerator[AsyncSession, None]:
    """Database session for direct DB operations in tests."""
    async with session_factory() as session:
        yield session
        await session.rollback()


@pytest_asyncio.fixture(scope="session")
async def app(engine: AsyncEngine, redis_client: aioredis.Redis):
    """FastAPI app with test engine injected (session-scoped for speed)."""
    reset_engine_for_testing(engine)
    app = create_app()
    # Block B1: contacts routes are plugin-owned and guarded by
    # require_active_plugin("contacts"). The test DB is empty, so create_app()
    # leaves active plugins empty — activate the core contacts plugin for the
    # generic app/client fixtures (same pattern as the specialized ai_app).
    from app.core.permission_registry import (
        init_permission_registry,
        register_plugin_permissions,
    )

    init_permission_registry(active_plugin_names={"contacts"})
    from app.plugins.builtins.contacts.plugin import ContactsPlugin

    contacts_manifest = ContactsPlugin().manifest
    if contacts_manifest.permissions:
        register_plugin_permissions("contacts", contacts_manifest.permissions)
    yield app
    await close_engine()


@pytest_asyncio.fixture
async def ai_app(engine: AsyncEngine, redis_client: aioredis.Redis):
    """FastAPI app with ai_assistant plugin activated for AI copilot tests."""
    from app.core.permission_registry import init_permission_registry, register_plugin_permissions
    reset_engine_for_testing(engine)
    app = create_app()
    # Re-initialize AFTER create_app() which reads active plugins from DB
    # (DB is empty in tests, so create_app leaves active_plugin_names empty)
    init_permission_registry(active_plugin_names={"ai_assistant"})
    # Register ai_assistant permissions so require_permission checks work
    from app.plugins.builtins.ai_assistant.plugin import AIAssistantPlugin
    plugin = AIAssistantPlugin()
    if hasattr(plugin.manifest, 'permissions') and plugin.manifest.permissions:
        register_plugin_permissions("ai_assistant", plugin.manifest.permissions)
    yield app
    await close_engine()


@pytest_asyncio.fixture
async def ai_client(ai_app) -> AsyncGenerator[AsyncClient, None]:
    """HTTP async test client with ai_assistant plugin active."""
    transport = ASGITransport(app=ai_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def client(app) -> AsyncGenerator[AsyncClient, None]:
    """HTTP async test client."""
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


# ─── Seed Data Helpers ───

ORIGIN_HEADER = {"Origin": "http://localhost:5173"}


async def seed_tenant_and_users(db: AsyncSession) -> dict[str, Any]:
    """Seed two tenants with admin, editor, viewer users.
    Returns dict with all created entity IDs.
    """
    tenant_a = Tenant(name="Tenant A", slug="tenant-a")
    tenant_b = Tenant(name="Tenant B", slug="tenant-b")
    db.add_all([tenant_a, tenant_b])
    await db.flush()

    # Admin in tenant A
    admin_a = User(
        email="admin@tenanta.com",
        name="Admin A",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    # Viewer in tenant A
    viewer_a = User(
        email="viewer@tenanta.com",
        name="Viewer A",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    # Editor in tenant A
    editor_a = User(
        email="editor@tenanta.com",
        name="Editor A",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    # Admin in tenant B
    admin_b = User(
        email="admin@tenantb.com",
        name="Admin B",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    db.add_all([admin_a, viewer_a, editor_a, admin_b])
    await db.flush()

    # Create admin role with *:* permissions for tenant A
    admin_role_a = Role(
        tenant_id=tenant_a.id,
        name="admin",
        permissions={"*": {"*": True}},
        denied_permissions=[],
        field_permissions={},
    )
    # Create viewer role for tenant A
    viewer_role_a = Role(
        tenant_id=tenant_a.id,
        name="viewer",
        permissions={"contacts": {"read": True}, "companies": {"read": True}, "calendar": {"read": True}, "dms": {"read": True}, "user_preferences": {"read": True, "write": True}},
        denied_permissions=[],
        field_permissions={},
    )
    # Create editor role for tenant A
    editor_role_a = Role(
        tenant_id=tenant_a.id,
        name="editor",
        permissions={"contacts": {"read": True, "write": True, "create": True, "update": True}, "companies": {"read": True, "write": True, "create": True, "update": True}},
        denied_permissions=[],
        field_permissions={},
    )
    # Create admin role for tenant B
    admin_role_b = Role(
        tenant_id=tenant_b.id,
        name="admin",
        permissions={"*": {"*": True}},
        denied_permissions=[],
        field_permissions={},
    )
    # Create a custom role with field permissions in tenant A
    custom_role = Role(
        tenant_id=tenant_a.id,
        name="sales_rep",
        permissions={"companies": {"read": True, "create": True, "update": True, "delete": False}, "contacts": {"read": True}},
        field_permissions={"annual_revenue": "hidden"},
    )
    db.add_all([admin_role_a, viewer_role_a, editor_role_a, admin_role_b, custom_role])
    await db.flush()

    # User-tenant memberships (with role_id linking to Role records)
    ut1 = UserTenant(user_id=admin_a.id, tenant_id=tenant_a.id, is_default=True, role="admin", role_id=admin_role_a.id)
    ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True, role="viewer", role_id=viewer_role_a.id)
    ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True, role="editor", role_id=editor_role_a.id)
    ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True, role="admin", role_id=admin_role_b.id)
    # Admin A is also member of tenant B (for switch-tenant test)
    ut5 = UserTenant(user_id=admin_a.id, tenant_id=tenant_b.id, is_default=False, role="admin", role_id=admin_role_b.id)
    db.add_all([ut1, ut2, ut3, ut4, ut5])
    await db.flush()

    # Create a company in tenant A
    company_a = Contact(
        tenant_id=tenant_a.id,
        type="company",
        name="Company Alpha",
        displayname="Company Alpha",
        created_by=admin_a.id,
        updated_by=admin_a.id,
    )
    # Create a company in tenant B
    company_b = Contact(
        tenant_id=tenant_b.id,
        type="company",
        name="Company Beta",
        displayname="Company Beta",
        created_by=admin_b.id,
        updated_by=admin_b.id,
    )
    db.add_all([company_a, company_b])
    await db.flush()

    await db.commit()

    return {
        "tenant_a": tenant_a,
        "tenant_b": tenant_b,
        "admin_a": admin_a,
        "viewer_a": viewer_a,
        "editor_a": editor_a,
        "admin_b": admin_b,
        "company_a": company_a,
        "company_b": company_b,
        "custom_role": custom_role,
        "admin_role_a": admin_role_a,
        "admin_role_b": admin_role_b,
    }


async def create_no_perm_user(db: AsyncSession, seed: dict[str, Any]) -> User:
    """Create a user in tenant A with no permissions (for RBAC tests).

    Returns the created user. The user has a role with empty permissions,
    so any require_permission check will return 403.
    """
    from app.core.auth import hash_password
    from app.models.role import Role
    from app.models.user import User, UserTenant

    no_perm_role = Role(
        tenant_id=seed["tenant_a"].id,
        name="no_perm",
        permissions={},
        denied_permissions=[],
        field_permissions={},
    )
    db.add(no_perm_role)
    await db.flush()

    no_perm_user = User(
        email="noperm@tenanta.com",
        name="No Perm",
        password_hash=hash_password("TestPass123!"),
        is_active=True,
        preferences={},
    )
    db.add(no_perm_user)
    await db.flush()

    ut = UserTenant(
        user_id=no_perm_user.id,
        tenant_id=seed["tenant_a"].id,
        is_default=True,
        role="no_perm",
        role_id=no_perm_role.id,
    )
    db.add(ut)
    await db.flush()
    await db.commit()
    return no_perm_user


async def login_client(
    client: AsyncClient, email: str, password: str = "TestPass123!"
) -> dict[str, str]:
    """Login via HTTP API, set CSRF token on client, return cookies dict."""
    resp = await client.post(
        "/api/v1/auth/login",
        json={"email": email, "password": password},
        headers=ORIGIN_HEADER,
    )
    assert resp.status_code == 200, f"Login failed: {resp.status_code} {resp.text}"
    data = resp.json()
    csrf_token = data.get("csrf_token", "")
    # Set csrf_token as default header on client (merged with per-request headers)
    client.headers["X-CSRF-Token"] = csrf_token
    # Also add Origin to client defaults so per-request headers aren't needed
    client.headers["Origin"] = ORIGIN_HEADER["Origin"]
    return dict(resp.cookies)


async def get_auth_client(
    client: AsyncClient, email: str, password: str = "TestPass123!"
) -> AsyncClient:
    """Return a client that's logged in."""
    await login_client(client, email, password)
    return client


DMS_TEST_STORAGE = "/tmp/dms_test"


@pytest_asyncio.fixture
async def dms_app(engine: AsyncEngine, redis_client):
    """FastAPI app with DMS + Permissions plugins registered."""
    os.environ["DMS_STORAGE_BASE"] = DMS_TEST_STORAGE
    reset_engine_for_testing(engine)
    app = create_app()

    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    init_permission_registry(active_plugin_names={"permissions", "dms", "tasks"})

    container = get_container()
    await container.initialize()

    from app.plugins.builtins.dms.plugin import DmsPlugin
    from app.plugins.builtins.permissions.plugin import PermissionsPlugin
    from app.plugins.builtins.tasks.plugin import TasksPlugin
    registry.register_plugin(PermissionsPlugin())
    registry.register_plugin(DmsPlugin())
    registry.register_plugin(TasksPlugin())
    reset_plugin_service_for_testing(registry)

    yield app
    await close_engine()
    # Cleanup test storage
    if os.path.exists(DMS_TEST_STORAGE):
        shutil.rmtree(DMS_TEST_STORAGE, ignore_errors=True)


@pytest_asyncio.fixture
async def dms_client(dms_app) -> AsyncClient:
    transport = ASGITransport(app=dms_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def authed_client(
    dms_client: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
    """Authenticated admin client with seeded data and both plugins activated."""
    seed = await seed_tenant_and_users(db_session)
    await login_client(dms_client, "admin@tenanta.com")

    # Install + activate permissions plugin first (DMS depends on it)
    resp = await dms_client.post("/api/v1/plugins/permissions/install", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"Permissions install failed: {resp.text}"
    resp = await dms_client.post("/api/v1/plugins/permissions/activate", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"Permissions activate failed: {resp.text}"

    # Install + activate DMS plugin
    resp = await dms_client.post("/api/v1/plugins/dms/install", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"DMS install failed: {resp.text}"
    resp = await dms_client.post("/api/v1/plugins/dms/activate", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"DMS activate failed: {resp.text}"

    return dms_client, seed


# ─── Calendar Fixtures ───


@pytest_asyncio.fixture
async def calendar_app(engine: AsyncEngine, redis_client):
    """FastAPI app with Calendar plugin registered, installed, and activated."""
    reset_engine_for_testing(engine)
    app = create_app()

    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    init_permission_registry(active_plugin_names={"calendar"})

    container = get_container()
    await container.initialize()

    registry.register_plugin(CalendarPlugin())
    reset_plugin_service_for_testing(registry)

    # Pre-install and activate the plugin so routes are registered
    # (tests that use viewer accounts can't install/activate — require_admin blocks them)
    _sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
    async with _sf() as session:
        await registry.install(session, "calendar")
        await registry.activate(session, "calendar")
        await session.commit()

    yield app
    await close_engine()


@pytest_asyncio.fixture
async def calendar_client(calendar_app) -> AsyncClient:
    transport = ASGITransport(app=calendar_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def calendar_authed_client(
    calendar_client: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
    """Authenticated admin client with seeded data and calendar plugin activated."""
    seed = await seed_tenant_and_users(db_session)
    await login_client(calendar_client, "admin@tenanta.com")

    # Install + activate calendar plugin
    resp = await calendar_client.post("/api/v1/plugins/calendar/install", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"Calendar install failed: {resp.text}"
    resp = await calendar_client.post("/api/v1/plugins/calendar/activate", headers=ORIGIN_HEADER)
    assert resp.status_code == 200, f"Calendar activate failed: {resp.text}"

    return calendar_client, seed


# ─── MCP Server / Client Fixtures ───────────────────────────────────────────

@pytest_asyncio.fixture
async def mcp_app(engine: AsyncEngine, redis_client):
    """FastAPI app with MCP Server + Client + Permissions plugins registered, installed, and activated."""
    reset_engine_for_testing(engine)
    app = create_app()

    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    init_permission_registry(active_plugin_names={"permissions", "mcp_server", "mcp_client"})

    container = get_container()
    await container.initialize()

    from app.plugins.builtins.mcp_client.plugin import McpClientPlugin
    from app.plugins.builtins.mcp_server.plugin import McpServerPlugin
    from app.plugins.builtins.permissions.plugin import PermissionsPlugin
    registry.register_plugin(PermissionsPlugin())
    registry.register_plugin(McpServerPlugin())
    registry.register_plugin(McpClientPlugin())
    reset_plugin_service_for_testing(registry)

    _sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
    async with _sf() as session:
        await registry.install(session, "permissions")
        await registry.activate(session, "permissions")
        await registry.install(session, "mcp_server")
        await registry.activate(session, "mcp_server")
        await registry.install(session, "mcp_client")
        await registry.activate(session, "mcp_client")
        await session.commit()

    yield app
    await close_engine()


@pytest_asyncio.fixture
async def mcp_client_fixture(mcp_app) -> AsyncClient:
    transport = ASGITransport(app=mcp_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c


@pytest_asyncio.fixture
async def mcp_authed_client(
    mcp_client_fixture: AsyncClient, db_session: AsyncSession
) -> tuple[AsyncClient, dict]:
    """Authenticated admin client with seeded data and MCP plugins activated."""
    seed = await seed_tenant_and_users(db_session)
    login_resp = await mcp_client_fixture.post(
        "/api/v1/auth/login",
        json={"email": "admin@tenanta.com", "password": "TestPass123!"},
        headers=ORIGIN_HEADER,
    )
    assert login_resp.status_code == 200, f"Login failed: {login_resp.text}"
    csrf_token = login_resp.json().get("csrf_token", "")
    mcp_client_fixture.headers.update({"X-CSRF-Token": csrf_token})

    return mcp_client_fixture, seed


# ─── Tasks Fixtures ──────────────────────────────────────────────────────────

@pytest_asyncio.fixture
async def tasks_app(engine: AsyncEngine, redis_client):
    """FastAPI app with Tasks + Permissions plugins registered, installed, and activated."""
    reset_engine_for_testing(engine)
    app = create_app()

    registry = reset_registry_for_testing()
    registry.initialize(engine, app)
    init_permission_registry(active_plugin_names={"permissions", "tasks"})

    container = get_container()
    await container.initialize()

    from app.plugins.builtins.permissions.plugin import PermissionsPlugin
    from app.plugins.builtins.tasks.plugin import TasksPlugin
    registry.register_plugin(PermissionsPlugin())
    registry.register_plugin(TasksPlugin())
    reset_plugin_service_for_testing(registry)

    _sf = async_sessionmaker(bind=engine, expire_on_commit=False, class_=AsyncSession)
    async with _sf() as session:
        await registry.install(session, "permissions")
        await registry.activate(session, "permissions")
        await registry.install(session, "tasks")
        await registry.activate(session, "tasks")
        await session.commit()

    yield app
    await close_engine()


@pytest_asyncio.fixture
async def tasks_client(tasks_app) -> AsyncClient:
    transport = ASGITransport(app=tasks_app)
    async with AsyncClient(transport=transport, base_url="http://test") as c:
        yield c