ctms/backend/tests/conftest.py

from __future__ import annotations

"""Pytest configuration and fixtures for tests."""

import asyncio
import uuid
from typing import AsyncGenerator

import pytest
import pytest_asyncio
from sqlalchemy import text, event, String, TypeDecorator
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.dialects.postgresql import UUID as PG_UUID


class GUID(TypeDecorator):
    """Platform-independent GUID type that uses CHAR(36) on SQLite."""
    impl = String
    cache_ok = True

    def load_dialect_impl(self, dialect):
        if dialect.name == 'sqlite':
            return dialect.type_descriptor(String(36))
        return dialect.type_descriptor(PG_UUID(as_uuid=True))

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        if dialect.name == 'sqlite':
            if isinstance(value, uuid.UUID):
                return str(value)
            return value
        return value

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        if dialect.name == 'sqlite':
            if isinstance(value, str):
                return uuid.UUID(value)
            return value
        return value


@pytest.fixture(scope="session")
def event_loop():
    """Create an instance of the default event loop for the test session."""
    loop = asyncio.new_event_loop()
    yield loop
    loop.close()


@pytest.fixture(scope="session")
def test_engine(event_loop):
    """Create a test database engine."""
    # Use SQLite in-memory database for testing
    engine = event_loop.run_until_complete(
        _create_test_engine()
    )
    yield engine
    event_loop.run_until_complete(engine.dispose())


async def _create_test_engine():
    """Helper to create and initialize test engine."""
    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        future=True,
        echo=False,
        connect_args={"check_same_thread": False},
    )

    # Import all models to register them with Base metadata
    from app.db.base_class import Base
    import app.models.study
    import app.models.study_role_permission
    import app.models.api_endpoint_permission
    import app.models.api_endpoint_registry
    import app.models.user
    import app.models.study_member
    import app.models.subject
    import app.models.visit
    import app.models.ae
    import app.models.monitoring_visit_issue
    import app.models.site

    # Replace PostgreSQL UUID type with custom GUID type for SQLite
    for table in Base.metadata.tables.values():
        for column in table.columns:
            if isinstance(column.type, PG_UUID):
                column.type = GUID()

    # Create all tables
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    return engine


@pytest_asyncio.fixture
async def db_session(test_engine) -> AsyncGenerator[AsyncSession, None]:
    """Create a test database session."""
    async_session = async_sessionmaker(
        bind=test_engine,
        class_=AsyncSession,
        expire_on_commit=False,
    )

    async with async_session() as session:
        # Create a test study for foreign key references with unique code
        study_code = f"TEST-STUDY-{uuid.uuid4().hex[:8]}"
        await session.execute(
            text("""
                INSERT INTO studies (id, code, name, status, is_locked, visit_schedule)
                VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule)
            """),
            {
                "id": str(uuid.uuid4()),
                "code": study_code,
                "name": "Test Study",
                "status": "ACTIVE",
                "is_locked": False,
                "visit_schedule": "[]",
            }
        )
        await session.commit()

        yield session

        # Clean up after test
        await session.rollback()