bbchops/transcription
1
0
Fork 0
generated from john/python-template
Code Issues 8 Pull Requests Actions Packages Projects Releases Wiki Activity

Revamped Sources related pages with the help of Gemini, which had a lot to say.

Browse Source
This commit is contained in:
Jim Lancaster
2026-08-05 12:35:54 -05:00
parent 4eeb552273
commit 72bc96ab3a
6 changed files with 416 additions and 298 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/transcription/db/models.py
+23
View File
@@ -181,6 +181,29 @@ class Source(SQLModel, table=True):
document: Optional["Document"] = Relationship(back_populates="sources", sa_relationship_kwargs={"lazy": "selectin"})
job_sources: list["JobSource"] = Relationship(back_populates="source", sa_relationship_kwargs={"lazy": "selectin"})
@property
def latest_job_source(self) -> Optional["JobSource"]:
"""Return the most recent job execution record for this source."""
if not self.job_sources:
return None
return max(self.job_sources, key=lambda js: js.executed_at)
@property
def latest_status(self) -> JobSourceStatus | None:
"""Return the execution status of the latest job run."""
latest = self.latest_job_source
return latest.status if latest else None
@property
def latest_error_detail(self) -> str | None:
"""Return the error detail from the latest job run, if present."""
latest = self.latest_job_source
return latest.error_detail if latest else None
@property
def document_name(self) -> str | None:
"""Return the parent document name if loaded."""
return self.document.name if self.document else None
class JobSource(SQLModel, table=True):
"""A single AI execution record for one source page."""
src/transcription/ui/components/table/common.py
+29 -19
View File
@@ -4,26 +4,21 @@ import logging
from collections.abc import Callable
from typing import Any
from nicegui import events
from nicegui import ui
from nicegui import events, ui
logger = logging.getLogger(__name__)
def _extract_row_id(args: Any) -> str | None:
if isinstance(args, dict):
if isinstance(args.get("row"), dict):
row_id = args["row"].get("id")
return str(row_id) if row_id is not None else None
row_id = args.get("id")
return str(row_id) if row_id is not None else None
if isinstance(args, list) and len(args) > 1 and isinstance(args[1], dict):
return str(args[1].get("id")) if args[1].get("id") is not None else None
if isinstance(args, list):
for value in args:
if isinstance(value, dict):
row_id = value.get("id")
if row_id is not None:
return str(row_id)
if isinstance(args, dict):
row = args.get("row")
if isinstance(row, dict) and "id" in row:
return str(row["id"])
if "id" in args:
return str(args["id"])
return None
@@ -40,7 +35,6 @@ def _bind_row_click_handler(
on_row_click_id(row_id)
table.on("rowClick", handle_row_click)
logger.debug("Row click handler bound to table")
def build_table(
@@ -49,15 +43,27 @@ def build_table(
*,
default_sort_by: str | None = None,
default_descending: bool = False,
classes: str = "app-table",
classes: str = "",
show_search: bool = True,
search_placeholder: str = "Search records...",
on_row_click_id: Callable[[str], None] | None = None,
) -> Any:
"""Build a styled Quasar table widget with optional client-side filtering and row-click handlers."""
pagination: dict[str, Any] = {"rowsPerPage": 25}
if default_sort_by is not None:
pagination["sortBy"] = default_sort_by
pagination["descending"] = default_descending
# Quasar props enforce behavior; visual styling is centralized in theme.css.
# Use a parent container to hold both the search bar and the table seamlessly
with ui.column().classes("w-full gap-2"):
if show_search:
with ui.row().classes("w-full items-center justify-end"):
search_input = (
ui.input(placeholder=search_placeholder)
.props("dense outlined clearable icon=search")
.classes("w-64 text-xs bg-white")
)
table = (
ui.table(
rows=rows,
@@ -65,7 +71,7 @@ def build_table(
row_key="id",
pagination=pagination,
)
.classes(f"w-full ui-table {classes}")
.classes(f"w-full ui-table {classes}".strip())
.props(
'flat square binary-state-sort table-style="table-layout: fixed; width: 100%;" '
'header-cell-class="ui-table-header text-xs uppercase tracking-wider" '
@@ -73,7 +79,11 @@ def build_table(
)
)
logger.info("Table built with %d rows and %d columns", len(rows), len(columns))
# Bind client-side text filter if search input is active
if show_search:
table.bind_filter_from(search_input, "value")
if on_row_click_id is not None:
_bind_row_click_handler(table, on_row_click_id=on_row_click_id)
return table
src/transcription/ui/components/table/sources.py
+51 -9
View File
@@ -23,6 +23,7 @@ class SourceTableRow:
upload_name: str
filename: str
document_id: UUID
document_name: str | None = None
job_source_status: str | None = None
job_source_error_detail: str | None = None
@@ -35,6 +36,7 @@ def _serialize_rows(rows: Sequence[SourceTableRow]) -> list[dict[str, Any]]:
"upload_name": row.upload_name,
"filename": row.filename,
"document_id": str(row.document_id),
"document_name": row.document_name or "-",
"job_source_status": row.job_source_status or "-",
"job_source_error_detail": row.job_source_error_detail or "-",
}
@@ -49,29 +51,69 @@ def render_sources_table(rows: Sequence[SourceTableRow]) -> None:
render_empty_state("No source file records found.")
return
build_table(
table = build_table(
rows=_serialize_rows(rows),
columns=[
{"name": "page_number", "label": "Page", "field": "page_number", "sortable": True},
{"name": "upload_name", "label": "Upload Title", "field": "upload_name", "sortable": True, "classes": "font-serif"},
{"name": "filename", "label": "Stored Filename", "field": "filename", "sortable": True, "classes": "font-mono"},
{
"name": "document_name",
"label": "Document Name",
"field": "document_name",
"sortable": True,
"classes": "font-serif",
},
{
"name": "page_number",
"label": "Page Number",
"field": "page_number",
"sortable": True,
},
{
"name": "upload_name",
"label": "Upload Title",
"field": "upload_name",
"sortable": True,
"classes": "font-serif",
},
{
"name": "filename",
"label": "Stored Filename",
"field": "filename",
"sortable": True,
"classes": "font-mono",
},
{
"name": "job_source_status",
"label": "Job Source Status",
"label": "Status",
"field": "job_source_status",
"sortable": True,
"classes": "font-mono",
},
{
"name": "job_source_error_detail",
"label": "Job Source Error Detail",
"label": "Error Detail",
"field": "job_source_error_detail",
"sortable": False,
"classes": "font-mono text-xs",
"classes": "font-mono text-xs truncate max-w-xs vibe-text-muted",
},
{"name": "document_id", "label": "Document ID", "field": "document_id", "sortable": True, "classes": "font-mono"},
],
default_sort_by="page_number",
classes="app-table w-full",
on_row_click_id=lambda source_id: ui.navigate.to(f"/sources/{source_id}"),
)
# Render job execution status using themed Quasar chips
table.add_slot(
"body-cell-job_source_status",
r"""
<q-td :props="props">
<q-chip
dense
square
size="sm"
:color="props.value === 'transcribed' ? 'positive' : props.value === 'failed' ? 'negative' : 'grey-5'"
text-color="white"
>
{{ props.value }}
</q-chip>
</q-td>
""",
)
src/transcription/ui/pages/sources_page.py
+145 -123
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
from typing import TYPE_CHECKING
from urllib.parse import urlencode
from uuid import UUID
@@ -9,7 +10,7 @@ from fastapi import Request
from fastapi.responses import RedirectResponse
from nicegui import ui
from transcription.db.models import JobSource, Source
from transcription.db.models import Source
from transcription.services.documents import DocumentError, DocumentService
from transcription.services.jobs import JobService
from transcription.services.transcription import (
@@ -22,14 +23,15 @@ from transcription.ui.components.cards import archival_card
from transcription.ui.components.data_display import metadata_row
from transcription.ui.components.document_panzoom import render_document_panzoom
from transcription.ui.components.error_presenter import show_error
from transcription.ui.components.primitives import destructive_button
from transcription.ui.components.primitives import section_header_row
from transcription.ui.components.primitives import destructive_button, section_header_row
from transcription.ui.components.table.sources import SourceTableRow, render_sources_table
from transcription.ui.theme import apply_archival_theme
from transcription.ui.theme import page_header
from transcription.ui.theme import apply_archival_theme, page_header
from ...db.session import SessionFactoryDep
if TYPE_CHECKING:
from starlette.datastructures import QueryParams
def register_page() -> None:
"""Register source list and detail routes."""
@@ -42,37 +44,34 @@ def register_page() -> None:
documents_service = DocumentService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
document_id_text = request.query_params.get("document_id")
job_id_text = request.query_params.get("job_id")
document_id = _parse_uuid(document_id_text)
job_id = _parse_uuid(job_id_text)
document_id = _parse_uuid(request.query_params.get("document_id"))
job_id = _parse_uuid(request.query_params.get("job_id"))
document_name = None
job_label = None
back_path = None
sources: list[Source] = []
job_source_by_source_id: dict[UUID, JobSource] = {}
try:
if document_id is not None:
document = await documents_service.read_document_detail(document_id=document_id)
document_name = document.name
back_path = f"/documents/{document.id}"
sources = list(sorted(document.sources, key=lambda item: (item.page_number, item.upload_name.casefold())))
sources = sorted(document.sources, key=lambda item: (item.page_number, item.upload_name.casefold()))
elif job_id is not None:
job = await jobs_service.read_job(job_id=job_id)
job_label = str(job.id)
back_path = f"/jobs/{job.id}"
job_sources = await sources_service.list_job_sources(job_id=job.id)
job_source_by_source_id = {
job_source.source_id: job_source
for job_source in job_sources
}
sources = [job_source.source for job_source in job_sources if job_source.source is not None]
sources.sort(key=lambda item: (item.page_number, item.upload_name.casefold()))
sources = sorted(
[js.source for js in job_sources if js.source is not None],
key=lambda item: (item.page_number, item.upload_name.casefold()),
)
else:
sources = list(sorted(await sources_service.list_sources(), key=lambda item: (item.document_id, item.page_number)))
sources = sorted(
await sources_service.list_sources(),
key=lambda item: (item.document_id, item.page_number),
)
except DocumentError:
ui.label("Document not found").classes("text-h6 text-red-800 p-4")
return
@@ -85,22 +84,18 @@ def register_page() -> None:
with ui.column().classes("w-full max-w-7xl mx-auto p-4 gap-4"):
with section_header_row():
if document_name is not None:
header_title = f"Sources: {document_name}"
elif job_label is not None:
header_title = f"Sources for Job {job_label}"
else:
header_title = "Archival Source Media"
header_title = _get_list_header_title(document_name, job_label)
page_header(header_title)
if back_path is not None:
back_label = "Back to Document" if document_id is not None else "Back to Job"
ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
"ui-btn-primary text-xs"
)
ui.button(
back_label,
on_click=lambda route=back_path: ui.navigate.to(route),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
# Format source records into read-model rows for the table renderer
# Clean row construction leveraging SQLModel @property definitions
rows = [
SourceTableRow(
id=source.id,
@@ -108,16 +103,9 @@ def register_page() -> None:
upload_name=source.upload_name,
filename=source.filename,
document_id=source.document_id,
job_source_status=(
job_source_by_source_id[source.id].status.value
if source.id in job_source_by_source_id
else None
),
job_source_error_detail=(
job_source_by_source_id[source.id].error_detail
if source.id in job_source_by_source_id
else None
),
document_name=source.document_name or document_name,
job_source_status=source.latest_status.value if source.latest_status else None,
job_source_error_detail=source.latest_error_detail,
)
for source in sources
]
@@ -129,9 +117,8 @@ def register_page() -> None:
sources_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
try:
parsed_source_id = UUID(source_id)
except ValueError:
parsed_source_id = _parse_uuid(source_id)
if parsed_source_id is None:
ui.label("Invalid source id").classes("text-h6 text-red-800 p-4")
return
@@ -149,30 +136,7 @@ def register_page() -> None:
with ui.column().classes("w-full max-w-[1800px] mx-auto p-4 gap-4"):
with section_header_row():
page_header(f"Source Page {source.page_number}: {source.upload_name}", subtitle=f"Source ID: {source.id}")
with ui.row().classes("items-center gap-2"):
if back_path is not None:
back_label = (
"Back to Document"
if "document_id" in request.query_params
else "Back to Job"
if "job_id" in request.query_params
else "Back to Sources"
)
ui.button(back_label, on_click=lambda route=back_path: ui.navigate.to(route), icon="arrow_back").classes(
"ui-btn-primary text-xs"
)
else:
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
"flat text-xs"
)
destructive_button(
"Delete Source",
on_click=lambda: ui.navigate.to(f"/sources/{source.id}/delete{_back_query(request.query_params)}"),
icon="delete",
extra_classes="text-xs",
)
_render_source_header_actions(source, back_path, request.query_params)
with ui.grid().classes("w-full grid-cols-12 gap-4"):
with ui.column().classes("col-span-12 lg:col-span-7 gap-4"):
@@ -180,57 +144,10 @@ def register_page() -> None:
render_document_panzoom(source=source)
with ui.column().classes("col-span-12 lg:col-span-5 gap-4"):
with archival_card(title="Source Metadata"):
metadata_row("Page Number:", str(source.page_number))
metadata_row("Upload Name:", source.upload_name)
metadata_row("Stored Filename:", source.filename)
metadata_row("Document ID:", str(source.document_id))
metadata_row("Date Uploaded:", source.date_uploaded.isoformat())
metadata_row(
"Date Revised:",
source.date_revised.isoformat() if source.date_revised else "Not revised",
)
with archival_card(title="Job Source Outcomes"):
if not source.job_sources:
ui.label("No job-source execution records found for this source.").classes("text-xs ui-text-muted")
else:
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
with ui.column().classes("w-full gap-1 p-2 ui-row-surface rounded"):
metadata_row("Job ID:", str(job_source.job_id))
metadata_row("Status:", job_source.status.value)
metadata_row("Executed At:", job_source.executed_at.isoformat())
metadata_row("Error Detail:", job_source.error_detail or "None")
with archival_card(title="Automated Raw Transcription"):
ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly bg-white").classes(
"w-full text-xs font-mono"
)
with archival_card(title="Curated Human Transcription"):
revision_input = (
ui.textarea(value=source.revised_text or "")
.props("outlined autogrow bg-white")
.classes("w-full text-xs")
)
async def save_revision() -> None:
candidate = (revision_input.value or "").strip()
if not candidate:
ui.notify("Revision text is required.", type="warning")
return
try:
await sources_service.upsert_revision_for_source(source_id=source.id, text=candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Save failed", operation="sources.save_revision")
return
ui.notify("Revision saved", type="positive")
ui.navigate.to(request.url.path + _back_query(request.query_params))
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save Revision", on_click=save_revision, icon="save").classes("ui-btn-primary text-xs")
_render_source_metadata(source)
_render_job_outcomes(source)
_render_raw_transcription(source)
_render_curated_transcription(source, sources_service, request)
@ui.page("/sources/{source_id}/delete")
async def source_delete_page(source_id: str, request: Request, session_factory: SessionFactoryDep) -> None:
@@ -238,9 +155,8 @@ def register_page() -> None:
sources_service = TranscriptionService(session_factory=session_factory)
render_navigation_header(current_path="/sources")
try:
parsed_source_id = UUID(source_id)
except ValueError:
parsed_source_id = _parse_uuid(source_id)
if parsed_source_id is None:
ui.label("Invalid source id").classes("text-h6 text-red-800 p-4")
return
@@ -308,6 +224,112 @@ def register_page() -> None:
return RedirectResponse(url=f"/ui/sources?job_id={job_id}")
# --- Component Extraction Helpers ---
def _render_source_header_actions(source: Source, back_path: str | None, query_params: QueryParams) -> None:
"""Render top actions for the source detail page."""
with ui.row().classes("items-center gap-2"):
if back_path is not None:
back_label = (
"Back to Document"
if "document_id" in query_params
else "Back to Job"
if "job_id" in query_params
else "Back to Sources"
)
ui.button(
back_label,
on_click=lambda route=back_path: ui.navigate.to(route),
icon="arrow_back",
).classes("ui-btn-primary text-xs")
else:
ui.button("Back to Sources", on_click=lambda: ui.navigate.to("/sources"), icon="arrow_back").props(
"flat text-xs"
)
destructive_button(
"Delete Source",
on_click=lambda: ui.navigate.to(f"/sources/{source.id}/delete{_back_query(query_params)}"),
icon="delete",
extra_classes="text-xs",
)
def _render_source_metadata(source: Source) -> None:
"""Render standard metadata fields for a source."""
with archival_card(title="Source Metadata"):
metadata_row("Page Number:", str(source.page_number))
metadata_row("Upload Name:", source.upload_name)
metadata_row("Stored Filename:", source.filename)
metadata_row("Document ID:", str(source.document_id))
metadata_row("Date Uploaded:", source.date_uploaded.isoformat())
metadata_row(
"Date Revised:",
source.date_revised.isoformat() if source.date_revised else "Not revised",
)
def _render_job_outcomes(source: Source) -> None:
"""Render related execution outcome cards."""
with archival_card(title="Job Source Outcomes"):
if not source.job_sources:
ui.label("No job-source execution records found for this source.").classes("text-xs ui-text-muted")
return
for job_source in sorted(source.job_sources, key=lambda item: item.executed_at, reverse=True):
with ui.column().classes("w-full gap-1 p-2 ui-row-surface rounded"):
metadata_row("Job ID:", str(job_source.job_id))
metadata_row("Status:", job_source.status.value)
metadata_row("Executed At:", job_source.executed_at.isoformat())
metadata_row("Error Detail:", job_source.error_detail or "None")
def _render_raw_transcription(source: Source) -> None:
"""Render raw machine transcription output."""
with archival_card(title="Automated Raw Transcription"):
ui.textarea(value=_source_transcription_text(source) or "").props("outlined autogrow readonly bg-white").classes(
"w-full text-xs font-mono"
)
def _render_curated_transcription(source: Source, sources_service: TranscriptionService, request: Request) -> None:
"""Render human revision editing panel."""
with archival_card(title="Curated Human Transcription"):
revision_input = (
ui.textarea(value=source.revised_text or "").props("outlined autogrow bg-white").classes("w-full text-xs")
)
async def save_revision() -> None:
candidate = (revision_input.value or "").strip()
if not candidate:
ui.notify("Revision text is required.", type="warning")
return
try:
await sources_service.upsert_revision_for_source(source_id=source.id, text=candidate)
except Exception as exc: # noqa: BLE001
show_error(exc, title="Save failed", operation="sources.save_revision")
return
ui.notify("Revision saved", type="positive")
ui.navigate.to(request.url.path + _back_query(request.query_params))
with ui.row().classes("w-full items-center gap-2 mt-2"):
ui.button("Save Revision", on_click=save_revision, icon="save").classes("ui-btn-primary text-xs")
# --- Utility Functions ---
def _get_list_header_title(document_name: str | None, job_label: str | None) -> str:
if document_name is not None:
return f"Sources: {document_name}"
if job_label is not None:
return f"Sources for Job {job_label}"
return "Archival Source Media"
def _parse_uuid(value: str | None) -> UUID | None:
if not value:
return None
@@ -330,7 +352,7 @@ def _source_detail_path(*, source_id: UUID, document_id: UUID | None, job_id: UU
return f"/sources/{source_id}{_build_filter_query(document_id=document_id, job_id=job_id)}"
def _back_query(query_params) -> str:
def _back_query(query_params: QueryParams) -> str:
params = {}
for key in ("document_id", "job_id"):
if query_params.get(key):
@@ -338,7 +360,7 @@ def _back_query(query_params) -> str:
return f"?{urlencode(params)}" if params else ""
def _back_path_from_query(query_params) -> str | None:
def _back_path_from_query(query_params: QueryParams) -> str | None:
document_id = query_params.get("document_id")
if document_id:
return f"/documents/{document_id}"
src/transcription/ui/static/theme.css
+15 -7
View File
@@ -229,26 +229,34 @@ input:focus-visible,
font-size: 0.75rem;
}
/* Table Archival Theme Bridge */
.ui-table {
border: 1px solid var(--theme-border);
color: var(--theme-text);
background: var(--theme-surface-raised);
border-radius: 0.125rem;
overflow: hidden;
}
.ui-table .q-table tbody tr:hover {
background: var(--theme-surface) !important;
cursor: pointer;
.ui-table .q-table {
color: var(--theme-text);
}
.ui-table .q-table th,
.ui-table-header {
color: var(--theme-inverse-text) !important;
background-color: var(--theme-primary) !important;
font-weight: 700;
}
.ui-table .q-table td {
border-bottom: 1px solid var(--theme-border) !important;
color: var(--theme-text);
}
.ui-table-header {
color: var(--theme-inverse-text);
background: var(--theme-primary);
font-weight: 700;
.ui-table .q-table tbody tr:hover {
background-color: var(--theme-surface) !important;
cursor: pointer;
}
.ui-table-body {
tests/ui/test_sources_page.py
+80 -67
View File
@@ -1,16 +1,60 @@
"""Tests for the sources page routes."""
"""Tests for the sources page routes and Source model properties."""
import asyncio
from pathlib import Path
import pytest
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from transcription.db import session_scope
from transcription.db.models import Document
from transcription.db.models import Job
from transcription.db.models import JobStatus
from transcription.db.models import Source
from transcription.db.models import Document, Job, JobSourceStatus, JobStatus, Source
# --- Unit Tests for Model @property Definitions ---
class TestSourceModelProperties:
"""Direct unit tests for Source computed properties."""
@pytest.mark.asyncio
async def test_source_properties_with_no_job_sources(self):
source = Source(
page_number=1,
upload_name="page_one.png",
filename="stored_page_one.png",
file_path="/tmp/stored_page_one.png",
)
assert source.latest_job_source is None
assert source.latest_status is None
assert source.latest_error_detail is None
assert source.document_name is None
@pytest.mark.asyncio
async def test_source_properties_with_document_and_job_sources(self, seed_job):
job_id = seed_job(
filename="source_prop_test.png",
status=JobStatus.FAILED,
transcription_text=None,
error_detail="Timeout during OCR parsing",
)
async with session_scope() as session:
job = await session.get(Job, job_id)
assert job is not None
source = (
await session.exec(select(Source).where(Source.document_id == job.document_id))
).first()
assert source is not None
# Validate computed properties
assert source.document_name is not None
assert source.latest_status == JobSourceStatus.FAILED
assert source.latest_error_detail == "Timeout during OCR parsing"
assert source.latest_job_source is not None
# --- Integration Tests for Page Rendering ---
@pytest.mark.integration
@@ -26,10 +70,10 @@ class TestSourcesPageRendering:
assert "Sources" in response.text
assert "No source file records found." in response.text
def test_sources_page_lists_seeded_sources(self, app_client):
@pytest.mark.asyncio
async def test_sources_page_lists_seeded_sources(self, app_client):
_, client = app_client
async def _seed() -> None:
async with session_scope() as session:
document = Document(name="Source Document", document_type="letter")
session.add(document)
@@ -45,23 +89,20 @@ class TestSourcesPageRendering:
)
await session.commit()
asyncio.run(_seed())
response = client.get("/ui/sources")
assert response.status_code == 200
assert "page_one.png" in response.text
assert "stored_page_one.png" in response.text
def test_sources_page_filters_to_document_context(self, app_client):
@pytest.mark.asyncio
async def test_sources_page_filters_to_document_context(self, app_client):
_, client = app_client
async def _seed() -> str:
async with session_scope() as session:
target = Document(name="Target", document_type="letter")
other = Document(name="Other", document_type="record")
session.add(target)
session.add(other)
session.add_all([target, other])
await session.flush()
session.add(
@@ -83,11 +124,9 @@ class TestSourcesPageRendering:
)
)
await session.commit()
return str(target.id)
target_id = str(target.id)
document_id = asyncio.run(_seed())
response = client.get(f"/ui/sources?document_id={document_id}")
response = client.get(f"/ui/sources?document_id={target_id}")
assert response.status_code == 200
assert "Sources: Target" in response.text
@@ -105,9 +144,10 @@ class TestSourcesPageRendering:
assert "Sources for Job" in response.text
assert "Back to Job" in response.text
assert "job-page.png" in response.text
assert "Job Source Status" in response.text
def test_sources_page_job_context_shows_job_source_status_and_error_detail(self, app_client, seed_job):
def test_sources_page_job_context_shows_job_source_status_and_error_detail(
self, app_client, seed_job
):
_, client = app_client
job_id = seed_job(
filename="job-failed-page.png",
@@ -123,9 +163,18 @@ class TestSourcesPageRendering:
assert "failed" in response.text.lower()
assert "Provider timed out" in response.text
def test_source_detail_page_renders_preview_and_revision_box(self, app_client, seed_job):
@pytest.mark.asyncio
async def test_source_detail_page_renders_preview_and_revision_box(
self, app_client, seed_job
):
_, client = app_client
fixture_path = Path(__file__).resolve().parents[1] / "fixtures" / "images" / "valid" / "small_png.png"
fixture_path = (
Path(__file__).resolve().parents[1]
/ "fixtures"
/ "images"
/ "valid"
/ "small_png.png"
)
job_id = seed_job(
filename="detail-source.png",
transcription_text="original transcription text",
@@ -133,7 +182,6 @@ class TestSourcesPageRendering:
source_file=fixture_path,
)
async def _get_source_id() -> str:
async with session_scope() as session:
job = await session.get(Job, job_id)
assert job is not None
@@ -141,57 +189,24 @@ class TestSourcesPageRendering:
await session.exec(select(Source).where(Source.document_id == job.document_id))
).first()
assert source is not None
return str(source.id)
source_id = asyncio.run(_get_source_id())
source_id = str(source.id)
response = client.get(f"/ui/sources/{source_id}")
assert response.status_code == 200
assert "Source Page 1: detail-source.png" in response.text
assert "Back to Sources" in response.text
assert "automated raw transcription" in response.text.lower()
assert "original transcription text" in response.text
assert "curated human transcription" in response.text.lower()
assert "human revision text" in response.text
assert "Page Number:" in response.text
assert "Stored Filename:" in response.text
assert "Delete Source" in response.text
def test_source_detail_page_displays_job_source_status_and_error_detail(self, app_client, seed_job):
_, client = app_client
job_id = seed_job(
filename="failed-source.png",
status=JobStatus.FAILED,
transcription_text=None,
error_detail="Provider timed out",
)
async def _get_source_id() -> str:
async with session_scope() as session:
job = await session.get(Job, job_id)
assert job is not None
source = (
await session.exec(select(Source).where(Source.document_id == job.document_id))
).first()
assert source is not None
return str(source.id)
source_id = asyncio.run(_get_source_id())
response = client.get(f"/ui/sources/{source_id}")
assert response.status_code == 200
assert "JOB SOURCE OUTCOMES" in response.text
assert "Status:" in response.text
assert "failed" in response.text.lower()
assert "Error Detail:" in response.text
assert "Provider timed out" in response.text
def test_source_delete_page_blocks_when_source_is_job_linked(self, app_client, seed_job):
@pytest.mark.asyncio
async def test_source_delete_page_blocks_when_source_is_job_linked(
self, app_client, seed_job
):
_, client = app_client
job_id = seed_job(filename="linked-source.png", transcription_text="linked text")
async def _get_source_id() -> str:
async with session_scope() as session:
job = await session.get(Job, job_id)
assert job is not None
@@ -199,19 +214,18 @@ class TestSourcesPageRendering:
await session.exec(select(Source).where(Source.document_id == job.document_id))
).first()
assert source is not None
return str(source.id)
source_id = str(source.id)
source_id = asyncio.run(_get_source_id())
response = client.get(f"/ui/sources/{source_id}/delete")
assert response.status_code == 200
assert "Delete Source Record" in response.text
assert "Delete is only available for unlinked sources." in response.text
def test_source_delete_page_allows_unlinked_source(self, app_client):
@pytest.mark.asyncio
async def test_source_delete_page_allows_unlinked_source(self, app_client):
_, client = app_client
async def _seed_unlinked_source() -> str:
async with session_scope() as session:
document = Document(name="Unlinked Source Doc", document_type="memo")
session.add(document)
@@ -225,9 +239,8 @@ class TestSourcesPageRendering:
)
session.add(source)
await session.commit()
return str(source.id)
source_id = str(source.id)
source_id = asyncio.run(_seed_unlinked_source())
response = client.get(f"/ui/sources/{source_id}/delete")
assert response.status_code == 200