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"

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.plugins.builtins.calendar import CalendarPlugin  # noqa: F401
from app.plugins.builtins.mcp_server import McpServerPlugin  # noqa: F401
from app.plugins.builtins.mcp_client import McpClientPlugin  # noqa: F401
from app.plugins.builtins.mcp_client.models import McpServerConfig  # noqa: F401
from app.plugins.builtins.calendar.models import (  # noqa: F401
    Calendar,
    CalendarEntry,
    CalendarEntryLink,
    CalendarShare,
    Resource,
    ResourceBooking,
    Subtask,
    UserCalendarVisibility,
)
from app.plugins.builtins.dms import DmsPlugin  # noqa: F401
from app.plugins.builtins.dms.models import File as DmsFile  # noqa: F401
from app.plugins.builtins.dms.models import Folder  # noqa: F401
from app.plugins.builtins.entity_links.models import EntityLink  # noqa: F401
from app.plugins.builtins.mail import MailPlugin  # noqa: F401
from app.plugins.builtins.mail.models import (  # noqa: F401
    ContactPgpKey,
    MailAccount,
    MailAccountDelegate,
    MailAccountSendPermission,
    MailAttachment,
    MailFolder,
    MailLabel,
    MailLabelAssignment,
    MailRule,
    MailSeenBy,
    MailSignature,
    MailTemplate,
    PgpKey,
    VacationSentLog,
)
from app.plugins.builtins.permissions import PermissionsPlugin  # noqa: F401
from app.plugins.builtins.permissions.models import Permission, ShareLink  # noqa: F401
from app.plugins.builtins.report_generator import ReportGeneratorPlugin  # noqa: F401
from app.plugins.builtins.report_generator.models import (  # noqa: F401
    ReportInstance,
    ReportTemplate,
)
from app.plugins.builtins.tags.models import Tag, TagAssignment  # noqa: F401
from app.plugins.builtins.tasks import TasksPlugin  # noqa: F401
from app.plugins.builtins.tasks.models import Task  # 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.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,
    )


@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:
        # 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())
    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
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
async def engine() -> AsyncGenerator[AsyncEngine, None]:
    """Async engine for the test database."""
    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
async def app(engine: AsyncEngine, redis_client: aioredis.Redis):
    """FastAPI app with test engine injected."""
    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()

    # User-tenant memberships
    ut1 = UserTenant(user_id=admin_a.id, tenant_id=tenant_a.id, is_default=True, role="admin")
    ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True, role="viewer")
    ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True, role="editor")
    ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True, role="admin")
    # 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")
    db.add_all([ut1, ut2, ut3, ut4, ut5])
    await db.flush()

    # 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}},
        field_permissions={"annual_revenue": "hidden"},
    )
    db.add(custom_role)
    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,
    }


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)

    container = get_container()
    await container.initialize()

    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)

    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)

    container = get_container()
    await container.initialize()

    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