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
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.company import Company
from app.models.contact import CompanyContact, Contact  # 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.workflow import Workflow, WorkflowInstance, WorkflowStepHistory  # noqa: F401
from app.plugins.builtins.calendar import CalendarPlugin  # 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.tags.models import Tag, TagAssignment  # 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"


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."""
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        # Drop all tables and types
        conn.execute(text("DROP SCHEMA public CASCADE;"))
        conn.execute(text("CREATE SCHEMA public;"))
        conn.execute(text("GRANT ALL ON SCHEMA public TO leocrm;"))
        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
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        conn.execute(text("DROP SCHEMA public CASCADE;"))
        conn.execute(text("CREATE SCHEMA public;"))
        conn.execute(text("GRANT ALL ON SCHEMA public TO leocrm;"))
        conn.commit()
    sync_eng.dispose()


@pytest.fixture(autouse=True)
def clean_tables(db_setup):
    """Clean all table data before each test (preserve schema)."""
    sync_eng = _get_sync_engine()
    with sync_eng.connect() as conn:
        # TRUNCATE all tables with CASCADE — fast and reliable isolation
        conn.execute(
            text(
                "TRUNCATE TABLE contact_pgp_keys, pgp_keys, mail_account_send_permissions, mail_account_delegates, mail_seen_by, vacation_sent_log, mail_signatures, mail_templates, mail_rules, mail_label_assignments, mail_labels, mail_attachments, mails, mail_folders, mail_accounts, resource_bookings, resources, subtasks, user_calendar_visibility, calendar_shares, calendar_entry_links, calendar_entries, calendars, files, folders, entity_links, share_links, permissions, tag_assignments, tags, workflow_step_history, workflow_instances, workflows, ai_messages, ai_conversations, plugin_migrations, plugins, company_contacts, contacts, api_tokens, password_reset_tokens, notifications, deletion_log, audit_log, sessions, roles, companies, user_tenants, users, tenants 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 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(
        tenant_id=tenant_a.id,
        email="admin@tenanta.com",
        name="Admin A",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        is_active=True,
        preferences={},
    )
    # Viewer in tenant A
    viewer_a = User(
        tenant_id=tenant_a.id,
        email="viewer@tenanta.com",
        name="Viewer A",
        password_hash=hash_password("TestPass123!"),
        role="viewer",
        is_active=True,
        preferences={},
    )
    # Editor in tenant A
    editor_a = User(
        tenant_id=tenant_a.id,
        email="editor@tenanta.com",
        name="Editor A",
        password_hash=hash_password("TestPass123!"),
        role="editor",
        is_active=True,
        preferences={},
    )
    # Admin in tenant B
    admin_b = User(
        tenant_id=tenant_b.id,
        email="admin@tenantb.com",
        name="Admin B",
        password_hash=hash_password("TestPass123!"),
        role="admin",
        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)
    ut2 = UserTenant(user_id=viewer_a.id, tenant_id=tenant_a.id, is_default=True)
    ut3 = UserTenant(user_id=editor_a.id, tenant_id=tenant_a.id, is_default=True)
    ut4 = UserTenant(user_id=admin_b.id, tenant_id=tenant_b.id, is_default=True)
    # 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)
    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 = Company(
        tenant_id=tenant_a.id,
        name="Company Alpha",
        industry="IT",
        created_by=admin_a.id,
        updated_by=admin_a.id,
    )
    # Create a company in tenant B
    company_b = Company(
        tenant_id=tenant_b.id,
        name="Company Beta",
        industry="Finance",
        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 and 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}"
    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())
    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