generated from john/python-template
ui test updates
This commit is contained in:
+51
-12
@@ -10,38 +10,63 @@ 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
|
||||
def app_client(tmp_path: Path) -> tuple[FastAPI, TestClient]:
|
||||
|
||||
@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(
|
||||
*,
|
||||
@@ -49,10 +74,17 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
|
||||
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:
|
||||
document = Document(filename=filename, file_path=f"uploads/{filename}")
|
||||
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()
|
||||
|
||||
@@ -60,16 +92,23 @@ def seed_job(app_client: tuple[FastAPI, TestClient]) -> Callable[..., UUID]:
|
||||
session.add(job)
|
||||
await session.flush()
|
||||
|
||||
if transcript_text is not None or error_detail is not None:
|
||||
session.add(
|
||||
Transcript(
|
||||
job_id=job.id,
|
||||
provider="openrouter",
|
||||
prompt_name="transcribe_document",
|
||||
text=transcript_text,
|
||||
error_detail=error_detail,
|
||||
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
|
||||
|
||||
+63
-39
@@ -1,53 +1,77 @@
|
||||
"""Tests for the jobs page route."""
|
||||
|
||||
from uuid import UUID
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from transcription.ui import register_pages
|
||||
from transcription.ui.pages import jobs_page
|
||||
from transcription.ui.pages.jobs_page import JobTableRow
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(monkeypatch):
|
||||
"""Provide a minimal app client with jobs data patched for rendering."""
|
||||
|
||||
async def _fetch_jobs_stub():
|
||||
return [
|
||||
JobTableRow(
|
||||
id=UUID("00000000-0000-0000-0000-000000000001"),
|
||||
status="queued",
|
||||
filename="sample.pdf",
|
||||
retry_count=2,
|
||||
created_at="2026-01-01T12:00:00+00:00",
|
||||
updated_at="2026-01-01T12:01:00+00:00",
|
||||
)
|
||||
]
|
||||
|
||||
monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_stub)
|
||||
|
||||
app = FastAPI()
|
||||
register_pages(app)
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
from transcription.models import JobStatus
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestPageRendering:
|
||||
"""Verify the jobs page is available and includes the main controls."""
|
||||
"""Verify jobs routes render correctly with real app wiring."""
|
||||
|
||||
def test_jobs_page_renders_expected_controls(self, client, monkeypatch):
|
||||
"""GET /ui/jobs returns the page shell and jobs controls."""
|
||||
response = client.get("/ui/jobs")
|
||||
|
||||
async def _fetch_jobs_empty():
|
||||
return []
|
||||
|
||||
monkeypatch.setattr(jobs_page, "fetch_jobs", _fetch_jobs_empty)
|
||||
def test_jobs_page_renders_empty_state(self, app_client):
|
||||
"""GET /ui/jobs renders the page and empty-state text when no jobs exist."""
|
||||
_, client = app_client
|
||||
response = client.get("/ui/jobs")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Transcription Jobs" in response.text
|
||||
assert "No jobs yet." in response.text
|
||||
|
||||
def test_jobs_page_lists_seeded_jobs(self, app_client, seed_job):
|
||||
"""GET /ui/jobs lists seeded jobs from the in-memory database."""
|
||||
_, client = app_client
|
||||
seed_job(filename="sample.pdf", status=JobStatus.TRANSCRIBED, transcript_text="done")
|
||||
|
||||
response = client.get("/ui/jobs")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "sample.pdf" in response.text
|
||||
assert "transcribed" in response.text
|
||||
|
||||
def test_job_detail_page_renders_seeded_job(self, app_client, seed_job):
|
||||
"""GET /ui/jobs/{job_id} renders detail content for a real seeded job."""
|
||||
_, client = app_client
|
||||
fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid" / "single_page_pdf.pdf"
|
||||
job_id = seed_job(
|
||||
filename="detail.pdf",
|
||||
status=JobStatus.TRANSCRIBED,
|
||||
transcript_revisions=[
|
||||
(0, None, "first attempt failed"),
|
||||
(1, "hello", None),
|
||||
],
|
||||
source_file=fixture_path,
|
||||
)
|
||||
|
||||
response = client.get(f"/ui/jobs/{job_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Job Detail" in response.text
|
||||
assert "Job overview" in response.text
|
||||
assert "detail.pdf" in response.text
|
||||
assert "Transcripts" in response.text
|
||||
assert "Revision" in response.text
|
||||
assert "first attempt failed" in response.text
|
||||
assert "hello" in response.text
|
||||
assert "Document preview" in response.text
|
||||
assert "/uploads/detail.pdf" in response.text
|
||||
|
||||
def test_job_detail_page_rejects_invalid_id(self, app_client):
|
||||
"""GET /ui/jobs/{job_id} shows validation feedback for malformed IDs."""
|
||||
_, client = app_client
|
||||
response = client.get("/ui/jobs/not-a-uuid")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Invalid job id" in response.text
|
||||
|
||||
def test_job_detail_page_handles_missing_job(self, app_client):
|
||||
"""GET /ui/jobs/{job_id} shows not-found state for unknown IDs."""
|
||||
_, client = app_client
|
||||
missing_id = uuid4()
|
||||
response = client.get(f"/ui/jobs/{missing_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Job not found" in response.text
|
||||
|
||||
@@ -1,39 +1,18 @@
|
||||
"""Tests for UI page registration wiring."""
|
||||
|
||||
import pytest
|
||||
from fastapi import FastAPI
|
||||
|
||||
from transcription.ui import register_pages
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestPageRegistration:
|
||||
"""Verify page registration and route wiring."""
|
||||
"""Verify page registration and mounted UI routes."""
|
||||
|
||||
def test_register_pages_wires_upload_jobs_and_mount(self, monkeypatch):
|
||||
"""register_pages registers pages and mounts NiceGUI at /ui."""
|
||||
calls: list[str] = []
|
||||
def test_ui_mount_serves_registered_pages(self, app_client):
|
||||
"""Mounted UI routes respond successfully when the full app is created."""
|
||||
_, client = app_client
|
||||
|
||||
def _record_upload() -> None:
|
||||
calls.append("upload")
|
||||
upload_response = client.get("/ui/upload")
|
||||
jobs_response = client.get("/ui/jobs")
|
||||
|
||||
def _record_jobs() -> None:
|
||||
calls.append("jobs")
|
||||
|
||||
def _record_run_with(
|
||||
_app: FastAPI,
|
||||
*,
|
||||
mount_path: str,
|
||||
show_welcome_message: bool,
|
||||
dark: bool,
|
||||
) -> None:
|
||||
calls.append(f"run_with:{mount_path}:{show_welcome_message}:{dark}")
|
||||
|
||||
monkeypatch.setattr("transcription.ui.register_upload_page", _record_upload)
|
||||
monkeypatch.setattr("transcription.ui.register_jobs_page", _record_jobs)
|
||||
monkeypatch.setattr("transcription.ui.ui.run_with", _record_run_with)
|
||||
|
||||
app = FastAPI()
|
||||
register_pages(app)
|
||||
|
||||
assert calls == ["upload", "jobs", "run_with:/ui:False:True"]
|
||||
assert upload_response.status_code == 200
|
||||
assert jobs_response.status_code == 200
|
||||
|
||||
@@ -1,55 +1,35 @@
|
||||
"""Tests for the upload page route."""
|
||||
|
||||
from pathlib import Path
|
||||
"""Tests for upload and entry-point routes."""
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from transcription.app import create_app
|
||||
from transcription.config import Settings
|
||||
from transcription.config import _settings
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path: Path):
|
||||
"""Provide a real app client backed by in-memory SQLite."""
|
||||
settings = Settings(
|
||||
openrouter_api_key="test-key",
|
||||
database_url="sqlite:///:memory:",
|
||||
environment="test",
|
||||
upload_dir=tmp_path / "uploads",
|
||||
prompt_dir=tmp_path / "prompts",
|
||||
)
|
||||
_settings.set(settings)
|
||||
|
||||
app = create_app()
|
||||
with TestClient(app) as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
class TestPageRendering:
|
||||
"""Verify the upload page is available and includes the main controls."""
|
||||
"""Verify upload-related routes return working pages."""
|
||||
|
||||
def test_root_redirects_to_ui(self, client):
|
||||
def test_root_redirects_to_ui(self, app_client):
|
||||
"""GET / redirects to the UI mount point."""
|
||||
_, client = app_client
|
||||
response = client.get("/", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 307
|
||||
assert response.headers["location"] == "/ui"
|
||||
|
||||
def test_ui_redirects_to_upload(self, client):
|
||||
def test_ui_redirects_to_upload(self, app_client):
|
||||
"""GET /ui redirects to the upload page."""
|
||||
_, client = app_client
|
||||
response = client.get("/ui", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 307
|
||||
assert response.headers["location"] == "/ui/upload"
|
||||
|
||||
def test_upload_page_renders_expected_controls(self, client):
|
||||
def test_upload_page_renders_expected_controls(self, app_client):
|
||||
"""GET /ui/upload returns the page shell and upload controls."""
|
||||
_, client = app_client
|
||||
response = client.get("/ui/upload")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "Upload Document" in response.text
|
||||
assert "Select document file" in response.text
|
||||
assert "Upload" in response.text
|
||||
assert "Jobs" in response.text
|
||||
|
||||
Reference in New Issue
Block a user