generated from john/python-template
429 lines
16 KiB
Python
429 lines
16 KiB
Python
"""Tests for the sources page routes and Source model properties."""
|
|
|
|
import re
|
|
from datetime import UTC
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
from PIL import Image
|
|
from sqlmodel import select
|
|
|
|
from transcription.db import session_scope
|
|
from transcription.db.loading import orm_attribute
|
|
from transcription.db.loading import selectinload
|
|
from transcription.db.models import Document
|
|
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 Source
|
|
from transcription.providers.evidence import TransportEvidence
|
|
from transcription.services.sources import SourceService
|
|
from transcription.ui.pages.sources_page import _is_wide_narrow_image
|
|
from transcription.ui.pages.sources_page import _resolve_source_detail_layout
|
|
|
|
# --- 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(
|
|
document_id=uuid4(),
|
|
page_number=1,
|
|
upload_name="page_one.png",
|
|
filename="stored_page_one.png",
|
|
file_path="/tmp/stored_page_one.png",
|
|
file_hash="b" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
|
|
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
|
|
|
|
def test_wide_narrow_aspect_detection_threshold(self):
|
|
assert _is_wide_narrow_image(width=2200, height=1000)
|
|
assert not _is_wide_narrow_image(width=1800, height=1000)
|
|
assert not _is_wide_narrow_image(width=0, height=1000)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_detail_layout_switches_for_wide_narrow_images(self, app_client):
|
|
app, _ = app_client
|
|
async with session_scope() as session:
|
|
document = Document(name="Layout test doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
|
|
wide_path = app.state.settings.upload_dir / "documents" / str(document.id) / "wide-strip.png"
|
|
wide_path.parent.mkdir(parents=True, exist_ok=True)
|
|
Image.new("RGB", (2400, 800), color=(255, 255, 255)).save(wide_path, format="PNG")
|
|
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="wide-strip.png",
|
|
filename="wide-strip.png",
|
|
file_path=wide_path.relative_to(app.state.settings.upload_dir).as_posix(),
|
|
file_hash="c" * 64,
|
|
file_size_bytes=wide_path.stat().st_size,
|
|
)
|
|
session.add(source)
|
|
await session.commit()
|
|
|
|
assert _resolve_source_detail_layout(source=source, settings=app.state.settings) == "wide_narrow"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_properties_with_document_and_job_sources(self, seed_job):
|
|
job_id = await 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)
|
|
.options(
|
|
selectinload(Source.document),
|
|
selectinload(Source.job_sources).selectinload(orm_attribute(JobSource.execution_attempts)),
|
|
)
|
|
.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
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_latest_error_detail_ignores_older_failures_when_latest_attempt_succeeds(self):
|
|
source = Source(
|
|
document_id=uuid4(),
|
|
page_number=1,
|
|
upload_name="page_one.png",
|
|
filename="stored_page_one.png",
|
|
file_path="/tmp/stored_page_one.png",
|
|
file_hash="b" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
job_source = JobSource(
|
|
job_id=uuid4(),
|
|
source_id=source.id,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
)
|
|
failed_attempt = ExecutionAttempt(
|
|
job_source_id=job_source.id,
|
|
job_id=job_source.job_id,
|
|
source_id=source.id,
|
|
attempt_number=1,
|
|
status=JobSourceStatus.FAILED,
|
|
provider="openrouter",
|
|
error_detail="Provider timeout",
|
|
started_at=datetime.now(UTC),
|
|
finished_at=datetime.now(UTC),
|
|
duration_ms=10,
|
|
)
|
|
successful_attempt = ExecutionAttempt(
|
|
job_source_id=job_source.id,
|
|
job_id=job_source.job_id,
|
|
source_id=source.id,
|
|
attempt_number=2,
|
|
status=JobSourceStatus.TRANSCRIBED,
|
|
provider="openrouter",
|
|
error_detail=None,
|
|
started_at=datetime.now(UTC),
|
|
finished_at=datetime.now(UTC),
|
|
duration_ms=10,
|
|
)
|
|
job_source.execution_attempts = [failed_attempt, successful_attempt]
|
|
source.job_sources = [job_source]
|
|
|
|
assert source.latest_error_detail is None
|
|
|
|
|
|
# --- Integration Tests for Page Rendering ---
|
|
|
|
|
|
@pytest.mark.integration
|
|
class TestSourcesPageRendering:
|
|
"""Verify source list and detail routes render expected states."""
|
|
|
|
def test_sources_page_renders_empty_state(self, app_client):
|
|
_, client = app_client
|
|
|
|
response = client.get("/ui/sources", follow_redirects=False)
|
|
|
|
assert response.status_code == 307
|
|
assert response.headers["location"] == "/ui/documents"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_page_lists_seeded_sources(self, app_client):
|
|
_, client = app_client
|
|
|
|
async with session_scope() as session:
|
|
document = Document(name="Source Document")
|
|
session.add(document)
|
|
await session.flush()
|
|
session.add(
|
|
Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="page_one.png",
|
|
filename="stored_page_one.png",
|
|
file_path="/tmp/stored_page_one.png",
|
|
file_hash="c" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
)
|
|
await session.commit()
|
|
document_id = str(document.id)
|
|
|
|
response = client.get(f"/ui/sources?document_id={document_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "Sources for Document" in response.text
|
|
assert "page_one.png" in response.text
|
|
assert "Source Document" in response.text
|
|
assert "Stored Filename" not in response.text
|
|
assert re.search(
|
|
r'"name":"upload_name","label":"Upload Title".*'
|
|
r'"name":"page_number","label":"Page Number".*'
|
|
r'"name":"document_name","label":"Document Name".*'
|
|
r'"name":"job_source_status","label":"Status".*'
|
|
r'"name":"job_source_error_detail","label":"Error Detail"',
|
|
response.text,
|
|
re.DOTALL,
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_page_filters_to_document_context(self, app_client):
|
|
_, client = app_client
|
|
|
|
async with session_scope() as session:
|
|
target = Document(name="Target")
|
|
other = Document(name="Other")
|
|
session.add_all([target, other])
|
|
await session.flush()
|
|
|
|
session.add_all(
|
|
[
|
|
Source(
|
|
document_id=target.id,
|
|
page_number=1,
|
|
upload_name="target_page.png",
|
|
filename="target_stored.png",
|
|
file_path="/tmp/target_stored.png",
|
|
file_hash="d" * 64,
|
|
file_size_bytes=1,
|
|
),
|
|
Source(
|
|
document_id=other.id,
|
|
page_number=1,
|
|
upload_name="other_page.png",
|
|
filename="other_stored.png",
|
|
file_path="/tmp/other_stored.png",
|
|
file_hash="e" * 64,
|
|
file_size_bytes=1,
|
|
),
|
|
]
|
|
)
|
|
await session.commit()
|
|
target_id = str(target.id)
|
|
|
|
response = client.get(f"/ui/sources?document_id={target_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "Sources for Document" in response.text
|
|
assert "target_page.png" in response.text
|
|
assert "other_page.png" not in response.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_page_filters_to_job_context(self, app_client, seed_job):
|
|
_, client = app_client
|
|
job_id = await seed_job(filename="job-page.png", transcription_text="job text")
|
|
|
|
response = client.get(f"/ui/sources?job_id={job_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "Sources for Job" in response.text
|
|
assert "job-page.png" in response.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sources_page_job_context_shows_job_source_status_and_error_detail(self, app_client, seed_job):
|
|
_, client = app_client
|
|
job_id = await seed_job(
|
|
filename="job-failed-page.png",
|
|
status=JobStatus.FAILED,
|
|
transcription_text=None,
|
|
error_detail="Provider timed out",
|
|
)
|
|
|
|
response = client.get(f"/ui/sources?job_id={job_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "job-failed-page.png" in response.text
|
|
assert "FAILED" in response.text
|
|
assert "Provider timed out" in response.text
|
|
|
|
@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"
|
|
job_id = await seed_job(
|
|
filename="detail-source.png",
|
|
transcription_text="original transcription text",
|
|
revision_text="human revision text",
|
|
source_file=fixture_path,
|
|
ai_metadata={"finish_reason": "stop", "confidence": 0.98},
|
|
raw_api_response={"id": "response-123", "model": "test-model"},
|
|
)
|
|
|
|
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
|
|
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.upper()
|
|
assert "SOURCE METADATA" in response.text.upper()
|
|
assert "SOURCEJOB METADATA" in response.text.upper()
|
|
assert "TRANSCRIPTION TEXT" not in response.text.upper()
|
|
assert "EDITABLE REVISION" in response.text.upper()
|
|
assert "human revision text" in response.text
|
|
assert "Save revision" in response.text
|
|
assert "Previous Page" in response.text
|
|
assert "Next Page" in response.text
|
|
assert "Normalized Metadata" in response.text
|
|
assert "OpenRouter SDK Response Snapshot" in response.text
|
|
assert "finish_reason" in response.text
|
|
assert "response-123" in response.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_detail_page_shows_unavailable_placeholder_for_unmanaged_media_path(self, app_client):
|
|
_, client = app_client
|
|
|
|
async with session_scope() as session:
|
|
document = Document(name="Missing media document")
|
|
session.add(document)
|
|
await session.flush()
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="missing-media.png",
|
|
filename="missing-media.png",
|
|
file_path=str((Path.cwd().parent / "external" / "missing-media.png").resolve()),
|
|
file_hash="f" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.commit()
|
|
source_id = str(source.id)
|
|
|
|
response = client.get(f"/ui/sources/{source_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "No source media available for inspection." in response.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_detail_separates_v42_evidence_layers(self, app_client, seed_job):
|
|
app, client = app_client
|
|
job_id = await seed_job(filename="evidence-source.png")
|
|
async with session_scope(session_factory=app.state.runtime.session_factory) 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
|
|
source_id = source.id
|
|
|
|
service = SourceService(session_factory=app.state.runtime.session_factory)
|
|
await service.update_job_source_transcription(
|
|
job_id=job_id,
|
|
source_id=source_id,
|
|
text="V4.2 transcription",
|
|
raw_api_response={"id": "sdk-snapshot"},
|
|
ai_metadata={"finish_reason": "stop"},
|
|
provider="openrouter",
|
|
model="vendor/model",
|
|
transport_evidence=TransportEvidence(
|
|
response_received=True,
|
|
status_code=200,
|
|
body=b'{"id":"transport-response"}',
|
|
safe_headers={"content-type": "application/json"},
|
|
content_type="application/json",
|
|
),
|
|
)
|
|
|
|
response = client.get(f"/ui/sources/{source_id}")
|
|
|
|
assert response.status_code == 200
|
|
assert "Export Evidence" in response.text
|
|
assert "Request Manifest" in response.text
|
|
assert "Transport Response" in response.text
|
|
assert "OpenRouter SDK Response Snapshot" in response.text
|
|
assert "Normalized Metadata" in response.text
|
|
assert "Software Context" in response.text
|
|
|
|
@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 = await seed_job(filename="linked-source.png", transcription_text="linked text")
|
|
|
|
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
|
|
source_id = str(source.id)
|
|
|
|
response = client.get(f"/ui/sources/{source_id}/delete")
|
|
|
|
assert response.status_code == 200
|
|
assert "DELETE SOURCE RECORD" in response.text.upper()
|
|
assert "Delete is only available for unlinked sources." in response.text
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_delete_page_allows_unlinked_source(self, app_client):
|
|
_, client = app_client
|
|
|
|
async with session_scope() as session:
|
|
document = Document(name="Unlinked Source Doc")
|
|
session.add(document)
|
|
await session.flush()
|
|
source = Source(
|
|
document_id=document.id,
|
|
page_number=1,
|
|
upload_name="orphan-source.png",
|
|
filename="orphan-source.png",
|
|
file_path="/tmp/orphan-source.png",
|
|
file_hash="f" * 64,
|
|
file_size_bytes=1,
|
|
)
|
|
session.add(source)
|
|
await session.commit()
|
|
source_id = str(source.id)
|
|
|
|
response = client.get(f"/ui/sources/{source_id}/delete")
|
|
|
|
assert response.status_code == 200
|
|
assert "DELETE SOURCE RECORD" in response.text.upper()
|
|
assert "Delete source permanently" in response.text
|
|
assert "Delete is only available for unlinked sources." not in response.text
|