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 runtime roles exist with hardened attributes (BUG-098)
            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; "
                "ELSE ALTER ROLE crm_api NOSUPERUSER NOBYPASSRLS; END IF; "
                "IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_worker') THEN "
                "CREATE ROLE crm_worker LOGIN PASSWORD 'crm_worker_password' NOSUPERUSER NOBYPASSRLS; "
                "ELSE ALTER ROLE crm_worker NOSUPERUSER NOBYPASSRLS; END IF; "
                "IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_migration') THEN "
                "ALTER ROLE crm_migration NOSUPERUSER NOBYPASSRLS; END IF; "
                "END $$;"
            ))

            # 2. Enable + FORCE RLS on all tenant tables. Exception: system
            #    identity tables stay RLS-free (BUG-098 vs bootstrap contract:
            #    login must read them WITHOUT tenant context — documented in
            #    cross_tenant v1 test_rls_disabled_on_system_tables).
            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.relname NOT IN ('alembic_version', 'users', 'user_tenants', 'groups', 'user_groups')
                          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);
                        EXECUTE format('ALTER TABLE public.%I FORCE ROW LEVEL SECURITY', r.tablename);
                    END LOOP;
                END $$;
            """))

            # 3. Create/replace tenant-isolation policies scoped to runtime roles
            #    (BUG-098: policies must be TO {crm_api, crm_worker}, not PUBLIC)
            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 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
                        -- Drop pre-existing policy of any scope, then recreate scoped
                        EXECUTE format('DROP POLICY IF EXISTS %I_tenant_isolation ON public.%I', r.tablename, r.tablename);
                        EXECUTE format(
                            'CREATE POLICY %I_tenant_isolation ON public.%I '
                            'FOR ALL TO crm_api, crm_worker '
                            '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. Drop legacy role if present (BUG-098) — never fatal: standard
            #    REASSIGN/DROP OWNED sequence; if dependencies remain we keep
            #    the role but strip login/privileges.
            conn.execute(text(
                "DO $$ BEGIN "
                "IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'crm_runtime') THEN "
                "BEGIN "
                "ALTER ROLE crm_runtime NOLOGIN; "
                "REASSIGN OWNED BY crm_runtime TO current_user; "
                "DROP OWNED BY crm_runtime; "
                "DROP ROLE crm_runtime; "
                "EXCEPTION WHEN OTHERS THEN "
                "ALTER ROLE crm_runtime NOLOGIN NOSUPERUSER NOBYPASSRLS; "
                "END; "
                "END IF; "
                "END $$;"
            ))

            # 5. Grant runtime roles access to all tables
            for _role in ("crm_api", "crm_worker"):
                conn.execute(text(f"GRANT USAGE ON SCHEMA public TO {_role}"))
                conn.execute(text(f"GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA public TO {_role}"))
                conn.execute(text(f"GRANT USAGE, SELECT ON ALL SEQUENCES IN SCHEMA public TO {_role}"))
            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.fixture(autouse=True)
def _reset_inmemory_rate_limiter():
    """Clear the process-local rate limiter around every test (BUG-097).

    The InMemoryRateLimiter is process-local and would otherwise accumulate
    password-reset/login attempts across tests, making later tests fail with
    429 depending purely on execution order.
    """
    from app.core.resilience import get_inmemory_limiter

    get_inmemory_limiter().clear_all()
    yield
    get_inmemory_limiter().clear_all()


@pytest_asyncio.fixture(autouse=True)
async def _clear_rate_limit_keys():
    """Delete Redis rate-limit keys after every test (BUG-097).

    check_rate_limit uses Redis INCR+EXPIRE keyed by client IP; all tests share
    the same test-client IP, so password-reset/login counters accumulate across
    tests and later tests hit 429 depending purely on execution order.

    Uses the SAME Redis connection the app uses (app settings REDIS_URL, DB 1 in
    .env.test) — the session-scoped redis_client fixture points at a different
    logical DB than get_redis() and would clean up nothing.
    """
    import redis.asyncio as aioredis_mod

    from app.config import get_settings

    url = get_settings().redis_url
    r = aioredis_mod.from_url(url, decode_responses=True)
    try:
        yield
        keys = await r.keys("rate:*")
        if keys:
            await r.delete(*keys)
    finally:
        await r.aclose()


@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