kite/backend/tests/conftest.py

"""
pytest configuration for DocuVault backend tests.

Plan 05 cutover: all sync flat-file fixtures (isolated_data_dir, sync client)
removed. Tests use async fixtures only.

Service availability detection:
  - INTEGRATION=1 env var: assume live Docker services are available
  - Default (no INTEGRATION): use in-memory SQLite + skip tests requiring real
    PostgreSQL/MinIO/Redis

SQLite compatibility note:
  The ORM models use PostgreSQL-specific types (UUID, INET, JSONB). SQLite does
  not understand these. The db_session fixture patches them before creating
  tables so the in-memory engine can build the schema successfully.
"""
from __future__ import annotations

import os
import socket

import pytest
import pytest_asyncio
from httpx import ASGITransport, AsyncClient
from sqlalchemy import String, Text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool


# ── Service availability ──────────────────────────────────────────────────────

def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
    """Return True if the given TCP port is reachable."""
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


@pytest.fixture(scope="session")
def live_services_available():
    """True when Docker Compose services are reachable (or INTEGRATION=1 is set)."""
    if os.environ.get("INTEGRATION") == "1":
        return True
    return (
        _port_open("localhost", 5432)
        and _port_open("localhost", 9000)
        and _port_open("localhost", 6379)
    )


# ── Core async fixtures ───────────────────────────────────────────────────────

def _patch_pg_types_for_sqlite():
    """Patch PostgreSQL-specific column types so SQLite can create the schema.

    SQLite does not know about INET, UUID (as_uuid=True), or JSONB. We
    replace them with Text/String equivalents for the in-memory test engine.
    This is done by monkey-patching the dialect-type mapping rather than
    modifying the models.
    """
    try:
        from sqlalchemy.dialects.postgresql import UUID as PG_UUID, INET, JSONB

        # Override compile methods so SQLite renders them as TEXT
        for pg_type in (INET, JSONB):
            pg_type.__class_getitem__ = classmethod(lambda cls, item: cls())

        # Patch impl so SQLite uses String
        if not hasattr(INET, "_sqlite_patched"):
            INET.impl = String
            INET._sqlite_patched = True
        if not hasattr(JSONB, "_sqlite_patched"):
            JSONB.impl = Text
            JSONB._sqlite_patched = True
    except Exception:
        pass  # If patching fails, the fixture will raise a CompileError naturally


@pytest_asyncio.fixture
async def db_session():
    """In-memory async SQLite session for unit tests.

    PostgreSQL-specific column types are overridden to Text/String so that
    Base.metadata.create_all works against the SQLite dialect.
    """
    from sqlalchemy.dialects.sqlite.base import SQLiteTypeCompiler
    from sqlalchemy.dialects.postgresql import INET, JSONB

    from db.models import Base

    # ── Type compatibility shims ──────────────────────────────────────────────
    # PostgreSQL-specific types (INET, JSONB) are unknown to the SQLite dialect.
    # Temporarily add visit_* methods that render them as TEXT so that
    # Base.metadata.create_all can build the schema in SQLite.

    _orig_visit_INET = getattr(SQLiteTypeCompiler, "visit_INET", None)
    _orig_visit_JSONB = getattr(SQLiteTypeCompiler, "visit_JSONB", None)

    def _visit_inet(self, type_, **kw):
        return "TEXT"

    def _visit_jsonb(self, type_, **kw):
        return "TEXT"

    SQLiteTypeCompiler.visit_INET = _visit_inet   # type: ignore[attr-defined]
    SQLiteTypeCompiler.visit_JSONB = _visit_jsonb  # type: ignore[attr-defined]

    # UUID(as_uuid=True) renders as CHAR(32) in SQLite — already handled by
    # SQLAlchemy's built-in UUID type mapping — no patch needed.

    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    try:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

        AsyncTestSession = async_sessionmaker(engine, expire_on_commit=False)
        async with AsyncTestSession() as session:
            yield session
    finally:
        await engine.dispose()
        # Restore compiler methods to leave no side effects on other tests
        if _orig_visit_INET is not None:
            SQLiteTypeCompiler.visit_INET = _orig_visit_INET  # type: ignore
        else:
            try:
                del SQLiteTypeCompiler.visit_INET  # type: ignore
            except AttributeError:
                pass
        if _orig_visit_JSONB is not None:
            SQLiteTypeCompiler.visit_JSONB = _orig_visit_JSONB  # type: ignore
        else:
            try:
                del SQLiteTypeCompiler.visit_JSONB  # type: ignore
            except AttributeError:
                pass


@pytest_asyncio.fixture
async def async_client(db_session: AsyncSession):
    """Async HTTP test client with the DB dependency overridden to use in-memory SQLite."""
    from deps.db import get_db
    from main import app

    app.dependency_overrides[get_db] = lambda: db_session

    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
        yield c

    app.dependency_overrides.clear()


# ── File fixtures ─────────────────────────────────────────────────────────────

@pytest.fixture
def sample_txt(tmp_path):
    p = tmp_path / "sample.txt"
    p.write_text("This is a test document about invoices and finance.")
    return p


@pytest.fixture
def sample_pdf(tmp_path):
    """Create a minimal valid PDF for testing."""
    import fitz

    doc = fitz.open()
    page = doc.new_page()
    page.insert_text((50, 50), "Test PDF document about contracts and legal matters.")
    pdf_path = tmp_path / "sample.pdf"
    doc.save(str(pdf_path))
    doc.close()
    return pdf_path