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 asyncio
import os
import shutil

# 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
from app.models.ai_conversation import AIConversation, AIMessage  # noqa: F401
from app.models.contact import Contact, ContactPerson  # noqa: F401
from app.models.contact_merge import ContactMergeHistory  # noqa: F401
from app.models.plugin import Plugin, PluginMigration  # noqa: F401
from app.models.role import Role
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.models.outbox import EventOutbox  # noqa: F401
from app.models.consumer_inbox import ConsumerInbox  # noqa: F401
from app.models.outbox_delivery import OutboxDelivery  # noqa: F401
from app.models.saved_filter import SavedFilter  # noqa: F401
from app.ai.oversight import DecisionRecordDB  # noqa: F401 — ensure table is created

# 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

from app.plugins.registry import reset_registry_for_testing  # noqa: F401
from app.core.permission_registry import init_permission_registry  # 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,
    )


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

    Uses SET lock_timeout to prevent deadlocks when multiple test processes
    try to DROP SCHEMA simultaneously. Falls back to TRUNCATE if DROP fails.
    """
    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 = '5s';"))
        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;"))
        except Exception:
            # If DROP SCHEMA deadlocks, fall back to TRUNCATE all tables
            conn.rollback()
            conn.execute(text("SET lock_timeout = '5s';"))
            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 $$;"
            ))
        conn.commit()
    sync_eng.dispose()

    # Create tables using async engine
    async def _create():
        eng = create_async_engine(TEST_DB_URL, echo=False)
        async with eng.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)
        await eng.dispose()

    asyncio.get_event_loop().run_until_complete(_create())

    # 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}")


    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:
        # Query existing table names from information_schema
        result = conn.execute(
            text(
                "SELECT table_name FROM information_schema.tables "
                "WHERE table_schema = 'public' AND table_type = 'BASE TABLE';"
            )
        )
        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()


@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()
    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.permissions.plugin import PermissionsPlugin
    from app.plugins.builtins.dms.plugin import DmsPlugin
    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.permissions.plugin import PermissionsPlugin
    from app.plugins.builtins.mcp_server.plugin import McpServerPlugin
    from app.plugins.builtins.mcp_client.plugin import McpClientPlugin
    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