generated from john/python-template
119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""Shared fixtures for UI integration tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlmodel import delete
|
|
|
|
from transcription.app import create_app
|
|
from transcription.config import Settings
|
|
from transcription.config import _settings
|
|
from transcription.db import create_all
|
|
from transcription.db import get_session
|
|
from transcription.db import initialize_database_runtime
|
|
from transcription.models import Document
|
|
from transcription.models import Job
|
|
from transcription.models import JobStatus
|
|
from transcription.models import Transcript
|
|
|
|
TranscriptSeed = tuple[int, str | None, str | None]
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def app_client(tmp_path_factory: pytest.TempPathFactory) -> tuple[FastAPI, TestClient]:
|
|
"""Provide a real application and test client backed by in-memory SQLite."""
|
|
tmp_path = tmp_path_factory.mktemp("ui")
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database_url="sqlite:///:memory:",
|
|
environment="test",
|
|
bootstrap_schema_on_startup=True,
|
|
upload_dir=tmp_path / "uploads",
|
|
prompt_dir=tmp_path / "prompts",
|
|
)
|
|
_settings.set(settings)
|
|
|
|
app = create_app()
|
|
app.state.runtime = initialize_database_runtime(settings=settings)
|
|
asyncio.run(create_all(engine=app.state.runtime.engine))
|
|
with TestClient(app) as client:
|
|
yield app, client
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_ui_database(app_client: tuple[FastAPI, TestClient]) -> None:
|
|
"""Reset UI-facing tables before each test for isolation."""
|
|
app, _ = app_client
|
|
|
|
async def _clear() -> None:
|
|
async with get_session(session_factory=app.state.runtime.session_factory) as session:
|
|
await session.exec(delete(Transcript))
|
|
await session.exec(delete(Job))
|
|
await session.exec(delete(Document))
|
|
await session.commit()
|
|
|
|
asyncio.run(_clear())
|
|
|
|
|
|
@pytest.fixture
|
|
def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
|
|
"""Return a helper for inserting a document/job/transcript trio."""
|
|
app, _ = app_client
|
|
fixtures_dir = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid"
|
|
|
|
def _seed(
|
|
*,
|
|
filename: str = "sample.pdf",
|
|
status: JobStatus = JobStatus.TRANSCRIBED,
|
|
transcript_text: str | None = "Sample transcript text",
|
|
error_detail: str | None = None,
|
|
transcript_revisions: list[TranscriptSeed] | None = None,
|
|
source_file: Path | None = None,
|
|
) -> UUID:
|
|
async def _insert() -> UUID:
|
|
async with get_session(session_factory=app.state.runtime.session_factory) as session:
|
|
stored_path = app.state.settings.upload_dir / filename
|
|
stored_path.parent.mkdir(parents=True, exist_ok=True)
|
|
source_path = source_file or fixtures_dir / "small_png.png"
|
|
stored_path.write_bytes(source_path.read_bytes())
|
|
|
|
document = Document(filename=filename, file_path=str(stored_path))
|
|
session.add(document)
|
|
await session.flush()
|
|
|
|
job = Job(document_id=document.id, status=status, retry_count=0)
|
|
session.add(job)
|
|
await session.flush()
|
|
|
|
revisions = transcript_revisions
|
|
if revisions is None and (transcript_text is not None or error_detail is not None):
|
|
revisions = [(0, transcript_text, error_detail)]
|
|
|
|
if revisions is not None:
|
|
for revision, revision_text, revision_error in revisions:
|
|
session.add(
|
|
Transcript(
|
|
job_id=job.id,
|
|
revision=revision,
|
|
provider="openrouter",
|
|
model="google/gemini-2.5-flash",
|
|
prompt_name="transcribe_document",
|
|
text=revision_text,
|
|
error_detail=revision_error,
|
|
)
|
|
)
|
|
|
|
await session.commit()
|
|
return job.id
|
|
|
|
return asyncio.run(_insert())
|
|
|
|
return _seed
|