Cheng Zhou
133 lines
3.9 KiB
Python
133 lines
3.9 KiB
Python
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.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
|
|
import app.models.permission_template
|
|
|
|
# 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, active_roles)
|
|
VALUES (:id, :code, :name, :status, :is_locked, :visit_schedule, :active_roles)
|
|
"""),
|
|
{
|
|
"id": str(uuid.uuid4()),
|
|
"code": study_code,
|
|
"name": "Test Study",
|
|
"status": "ACTIVE",
|
|
"is_locked": False,
|
|
"visit_schedule": "[]",
|
|
"active_roles": "[]",
|
|
}
|
|
)
|
|
await session.commit()
|
|
|
|
yield session
|
|
|
|
# Clean up after test
|
|
await session.rollback()
|