generated from john/python-template
175 lines
6.6 KiB
Python
175 lines
6.6 KiB
Python
"""Shared fixtures for UI integration tests."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Awaitable
|
|
from collections.abc import Callable
|
|
from collections.abc import Generator
|
|
from datetime import UTC
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
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 SqliteSettings
|
|
from transcription.db import session as db_session_module
|
|
from transcription.db import session_scope
|
|
from transcription.db.models import Document
|
|
from transcription.db.models import DocumentPerson
|
|
from transcription.db.models import DocumentTag
|
|
from transcription.db.models import ExecutionAttempt
|
|
from transcription.db.models import Job
|
|
from transcription.db.models import JobSource
|
|
from transcription.db.models import JobSourceStatus
|
|
from transcription.db.models import JobStatus
|
|
from transcription.db.models import Person
|
|
from transcription.db.models import Source
|
|
from transcription.db.models import Tag
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def app_client(tmp_path_factory: pytest.TempPathFactory) -> Generator[tuple[FastAPI, TestClient]]:
|
|
"""Provide a real application and test client backed by in-memory SQLite."""
|
|
tmp_path = tmp_path_factory.mktemp("ui")
|
|
database = SqliteSettings(path=str(tmp_path / "ui-tests.db"))
|
|
settings = Settings(
|
|
openrouter_api_key="test-key",
|
|
database=database,
|
|
environment="test",
|
|
bootstrap_schema_on_startup=True,
|
|
upload_dir=tmp_path / "uploads",
|
|
prompt_dir=tmp_path / "prompts",
|
|
)
|
|
|
|
app = create_app(settings=settings)
|
|
|
|
with TestClient(app) as client:
|
|
runtime_path = Path(str(app.state.runtime.engine.url.database)).resolve()
|
|
expected_path = Path(database.path).resolve()
|
|
if runtime_path != expected_path:
|
|
raise RuntimeError(
|
|
"Refusing to initialize destructive UI fixtures against "
|
|
f"{runtime_path}; expected {expected_path}"
|
|
)
|
|
yield app, client
|
|
|
|
|
|
@pytest_asyncio.fixture(autouse=True)
|
|
async def clear_ui_database(
|
|
app_client: tuple[FastAPI, TestClient],
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Reset UI-facing tables asynchronously before each test for isolation."""
|
|
app, _ = app_client
|
|
monkeypatch.setattr(
|
|
db_session_module,
|
|
"resolve_session_factory",
|
|
lambda *_args, **_kwargs: app.state.runtime.session_factory,
|
|
)
|
|
async with session_scope(session_factory=app.state.runtime.session_factory) as session:
|
|
await session.exec(delete(JobSource))
|
|
await session.exec(delete(DocumentTag))
|
|
await session.exec(delete(DocumentPerson))
|
|
await session.exec(delete(Source))
|
|
await session.exec(delete(Job))
|
|
await session.exec(delete(Document))
|
|
await session.exec(delete(Person))
|
|
await session.exec(delete(Tag))
|
|
await session.commit()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., Awaitable[UUID]]:
|
|
"""Return an async helper for seeding a Document -> Job -> Source tuple."""
|
|
app, _ = app_client
|
|
fixtures_dir = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid"
|
|
|
|
async def _seed(
|
|
*,
|
|
filename: str = "sample.pdf",
|
|
status: JobStatus = JobStatus.TRANSCRIBED,
|
|
transcription_text: str | None = "Sample transcript text",
|
|
error_detail: str | None = None,
|
|
revision_text: str | None = None,
|
|
source_file: Path | None = None,
|
|
ai_metadata: dict | None = None,
|
|
raw_api_response: dict | None = None,
|
|
) -> UUID:
|
|
async with session_scope() 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(name=filename)
|
|
session.add(document)
|
|
await session.flush()
|
|
|
|
job = Job(
|
|
document_id=document.id,
|
|
status=status,
|
|
retry_count=0,
|
|
provider="openrouter",
|
|
model="google/gemini-2.5-flash",
|
|
prompt_name="transcribe_document.md",
|
|
)
|
|
session.add(job)
|
|
await session.flush()
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name=filename,
|
|
filename=filename,
|
|
file_path=stored_path.resolve().relative_to(app.state.settings.upload_dir.resolve()).as_posix(),
|
|
file_hash="b" * 64,
|
|
file_size_bytes=len(stored_path.read_bytes()),
|
|
)
|
|
session.add(source)
|
|
await session.flush()
|
|
|
|
if transcription_text is not None or error_detail is not None:
|
|
outcome = JobSourceStatus.TRANSCRIBED if transcription_text is not None else JobSourceStatus.FAILED
|
|
job_source = JobSource(job_id=job.id, source_id=source.id, status=outcome)
|
|
session.add(job_source)
|
|
await session.flush()
|
|
|
|
# V4.7: evidence lives on execution_attempt, not job_source.
|
|
executed_at = datetime.now(UTC)
|
|
session.add(
|
|
ExecutionAttempt(
|
|
job_source_id=job_source.id,
|
|
job_id=job.id,
|
|
source_id=source.id,
|
|
attempt_number=1,
|
|
status=outcome,
|
|
provider="openrouter",
|
|
model="google/gemini-2.5-flash",
|
|
response_received=transcription_text is not None,
|
|
sdk_response_snapshot=raw_api_response,
|
|
normalized_metadata=ai_metadata,
|
|
raw_transcription=transcription_text,
|
|
error_detail=error_detail,
|
|
started_at=executed_at,
|
|
finished_at=executed_at,
|
|
duration_ms=0,
|
|
)
|
|
)
|
|
|
|
if revision_text is not None:
|
|
source.revised_text = revision_text
|
|
source.date_revised = datetime.now(UTC)
|
|
session.add(source)
|
|
|
|
await session.commit()
|
|
return job.id
|
|
|
|
return _seed
|