"""Tests for the sources page routes and Source model properties.""" from pathlib import Path import pytest from sqlmodel import select from transcription.db import session_scope from transcription.db.models import Document from transcription.db.models import Job 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 # --- 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", 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 @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).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 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") assert response.status_code == 200 assert "Source Asset Records" in response.text assert "No source asset records found in repository." in response.text assert "Upload New Documents" not in response.text @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", document_type="letter") 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() response = client.get("/ui/sources") assert response.status_code == 200 assert "page_one.png" in response.text assert "Source Document" in response.text assert "Stored Filename" not in response.text @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", document_type="letter") other = Document(name="Other", document_type="record") 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.lower() 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" in response.text.upper() assert "EDITABLE REVISION" in response.text.upper() assert "original transcription text" in response.text 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 "AI Metadata" in response.text assert "Raw API Response" in response.text assert "finish_reason" in response.text assert "response-123" 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 assert "Derived Artifacts" 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", document_type="memo") 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