- Fix UV index strategy: mark PyTorch CPU index as explicit with name - Add --index-strategy unsafe-best-match to Dockerfile uv pip install - Fix redis version constraint (>=5.0,<6) for ARQ compatibility - Fix Anthropic model name (claude-sonnet-4-5-20250929) - Fix IMAGE_PROVIDER enum value (gemini_flash, not google) - Resolve middlewares.py vs middlewares/ package conflict - Fix worker import paths (models.sql.presentation, models.sql.slide, utils split) - Fix seed script FK resolution by importing all related models - Fix test suite: async fixture scoping, greenlet dep, regex patterns, fixture params - Fix frontend TypeScript error (Boolean cast for layout.react_code) - Regenerate package-lock.json with i18n packages - Add initial Alembic migration (autogenerated from all models) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
215 lines
6.8 KiB
Python
215 lines
6.8 KiB
Python
"""Test fixtures: async SQLite database, factory functions for models."""
|
|
import os
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
|
|
from sqlmodel import SQLModel
|
|
|
|
# Force SQLite for tests (must be set before any app imports)
|
|
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///:memory:")
|
|
os.environ.setdefault("JWT_SECRET_KEY", "test-secret-key")
|
|
os.environ.setdefault("DEV_AUTH_PASSWORD", "testpass")
|
|
|
|
# Import all SQLModel table classes so metadata is populated
|
|
from models.sql.user import UserModel # noqa: E402
|
|
from models.sql.client import ClientModel # noqa: E402
|
|
from models.sql.team import TeamModel # noqa: E402
|
|
from models.sql.team_membership import TeamMembershipModel # noqa: E402
|
|
from models.sql.brand_config import BrandConfigModel # noqa: E402
|
|
from models.sql.audit_log import AuditLogModel # noqa: E402
|
|
from models.sql.presentation import PresentationModel # noqa: E402
|
|
from models.sql.job import JobModel # noqa: E402
|
|
from models.sql.master_deck import MasterDeckModel # noqa: E402
|
|
|
|
|
|
_test_engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
|
_test_session_maker = async_sessionmaker(_test_engine, expire_on_commit=False)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def _setup_database():
|
|
"""Create all tables before each test, drop after."""
|
|
async with _test_engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
yield
|
|
async with _test_engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.drop_all)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def session(_setup_database) -> AsyncSession:
|
|
async with _test_session_maker() as s:
|
|
yield s
|
|
|
|
|
|
# --------------- Factory Fixtures ---------------
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_user(session: AsyncSession):
|
|
async def _factory(
|
|
email: str = "test@oliver.com",
|
|
role: str = "user",
|
|
display_name: str = "Test User",
|
|
is_active: bool = True,
|
|
azure_oid: str | None = None,
|
|
) -> UserModel:
|
|
user = UserModel(
|
|
email=email,
|
|
display_name=display_name,
|
|
role=role,
|
|
is_active=is_active,
|
|
azure_oid=azure_oid,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(user)
|
|
await session.commit()
|
|
await session.refresh(user)
|
|
return user
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_client(session: AsyncSession):
|
|
async def _factory(
|
|
name: str = "Acme Corp",
|
|
slug: str | None = None,
|
|
retention_days: int | None = None,
|
|
is_active: bool = True,
|
|
) -> ClientModel:
|
|
client = ClientModel(
|
|
name=name,
|
|
slug=slug or name.lower().replace(" ", "-"),
|
|
retention_days=retention_days,
|
|
is_active=is_active,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(client)
|
|
await session.commit()
|
|
await session.refresh(client)
|
|
return client
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_team(session: AsyncSession):
|
|
async def _factory(
|
|
name: str = "Dev Team",
|
|
client_id: uuid.UUID | None = None,
|
|
is_default: bool = False,
|
|
) -> TeamModel:
|
|
team = TeamModel(
|
|
name=name,
|
|
client_id=client_id,
|
|
is_default=is_default,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(team)
|
|
await session.commit()
|
|
await session.refresh(team)
|
|
return team
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_membership(session: AsyncSession):
|
|
async def _factory(user_id: uuid.UUID, team_id: uuid.UUID) -> TeamMembershipModel:
|
|
membership = TeamMembershipModel(
|
|
user_id=user_id,
|
|
team_id=team_id,
|
|
assigned_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(membership)
|
|
await session.commit()
|
|
await session.refresh(membership)
|
|
return membership
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_brand_config(session: AsyncSession):
|
|
async def _factory(
|
|
client_id: uuid.UUID,
|
|
primary_colors: list | None = None,
|
|
secondary_colors: list | None = None,
|
|
fonts: dict | None = None,
|
|
logo_paths: list | None = None,
|
|
voice_rules: str | None = None,
|
|
voice_examples: list | None = None,
|
|
) -> BrandConfigModel:
|
|
brand = BrandConfigModel(
|
|
client_id=client_id,
|
|
primary_colors=primary_colors or ["#5146E5", "#3D35B0"],
|
|
secondary_colors=secondary_colors or ["#E9E8F8"],
|
|
fonts=fonts or {"heading": "Inter", "body": "Roboto"},
|
|
logo_paths=logo_paths,
|
|
voice_rules=voice_rules,
|
|
voice_examples=voice_examples,
|
|
created_at=datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(brand)
|
|
await session.commit()
|
|
await session.refresh(brand)
|
|
return brand
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_presentation(session: AsyncSession):
|
|
async def _factory(
|
|
owner_id: uuid.UUID | None = None,
|
|
client_id: uuid.UUID | None = None,
|
|
title: str = "Test Deck",
|
|
status: str = "draft",
|
|
created_at: datetime | None = None,
|
|
deleted_at: datetime | None = None,
|
|
) -> PresentationModel:
|
|
pres = PresentationModel(
|
|
content="Test content",
|
|
n_slides=5,
|
|
language="English",
|
|
title=title,
|
|
owner_id=owner_id,
|
|
client_id=client_id,
|
|
status=status,
|
|
created_at=created_at or datetime.now(timezone.utc),
|
|
updated_at=datetime.now(timezone.utc),
|
|
deleted_at=deleted_at,
|
|
)
|
|
session.add(pres)
|
|
await session.commit()
|
|
await session.refresh(pres)
|
|
return pres
|
|
return _factory
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def make_job(session: AsyncSession):
|
|
async def _factory(
|
|
user_id: uuid.UUID,
|
|
client_id: uuid.UUID,
|
|
job_type: str = "generate_presentation",
|
|
status: str = "queued",
|
|
progress: int = 0,
|
|
presentation_id: uuid.UUID | None = None,
|
|
) -> JobModel:
|
|
job = JobModel(
|
|
user_id=user_id,
|
|
client_id=client_id,
|
|
job_type=job_type,
|
|
status=status,
|
|
progress=progress,
|
|
presentation_id=presentation_id,
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(job)
|
|
await session.commit()
|
|
await session.refresh(job)
|
|
return job
|
|
return _factory
|