"""Tests for the jobs page routes and action handlers.""" import pytest import pytest_asyncio from transcription.db import session_scope from transcription.db.models import Document from transcription.db.models import Job from transcription.db.models import JobStatus # --- Helper Fixtures --- @pytest_asyncio.fixture async def seed_document_with_unlinked_job(): """Seed a document and a queued job for testing route actions.""" async with session_scope() as session: document = Document(name="Test Archival Letter") session.add(document) await session.flush() job = Job( document_id=document.id, status=JobStatus.QUEUED, provider="openai", model="gpt-4o", ) session.add(job) await session.commit() return str(document.id), str(job.id) # --- Integration Tests for Jobs Route Handlers --- @pytest.mark.integration class TestJobsPageRendering: """Verify jobs list, creation, detail, and lifecycle action routes.""" def test_jobs_page_renders_empty_state(self, app_client): _, client = app_client response = client.get("/ui/jobs") assert response.status_code == 200 assert "Transcription Pipeline Jobs" in response.text assert "No job records found in repository." in response.text @pytest.mark.asyncio async def test_jobs_page_lists_seeded_jobs(self, app_client, seed_job): _, client = app_client job_id = await seed_job(filename="seeded-document-page.png") response = client.get("/ui/jobs") assert response.status_code == 200 assert "Document Name" in response.text assert "# Sources" in response.text assert "Source Filename" not in response.text assert "Updated" in response.text assert "Created" not in response.text assert str(job_id) in response.text @pytest.mark.asyncio async def test_jobs_page_filters_for_document_context(self, app_client): _, client = app_client async with session_scope() as session: first = Document(name="First Doc") second = Document(name="Second Doc") session.add_all([first, second]) await session.flush() first_job = Job(document_id=first.id, status=JobStatus.QUEUED, provider="openai", model="gpt-4o") second_job = Job(document_id=second.id, status=JobStatus.QUEUED, provider="openai", model="gpt-4o") session.add_all([first_job, second_job]) await session.commit() first_id = str(first.id) first_job_id = str(first_job.id) second_job_id = str(second_job.id) response = client.get(f"/ui/jobs?document_id={first_id}") assert response.status_code == 200 assert "Jobs for Document" in response.text assert "Create job" not in response.text assert "Refresh" not in response.text assert first_job_id in response.text assert second_job_id not in response.text @pytest.mark.asyncio async def test_job_create_page_shows_empty_document_warning_when_no_docs(self, app_client): _, client = app_client response = client.get("/ui/jobs/new") assert response.status_code == 200 assert "Create Processing Job" in response.text assert "No documents available. Create a Document before creating a Job." in response.text @pytest.mark.asyncio async def test_job_create_page_preselects_document_from_query_param(self, app_client): _, client = app_client async with session_scope() as session: doc = Document(name="Preselected Journal Entry") session.add(doc) await session.commit() doc_id = str(doc.id) response = client.get(f"/ui/jobs/new?document_id={doc_id}") assert response.status_code == 200 assert "Create Processing Job" in response.text assert "Preselected Journal Entry" in response.text assert "Provider" in response.text assert "Model" in response.text @pytest.mark.asyncio async def test_job_detail_page_renders_logistics_and_links(self, app_client, seed_document_with_unlinked_job): _, client = app_client _, job_id = seed_document_with_unlinked_job response = client.get(f"/ui/jobs/{job_id}") assert response.status_code == 200 assert f"Job Record: {job_id}" in response.text assert "Execution Logistics".upper() in response.text.upper() assert "openai" in response.text assert "gpt-4o" in response.text assert "Document Name:" in response.text assert "Sources:" in response.text assert "View Sources" in response.text assert "View Linked Document" not in response.text assert "View Linked Sources" not in response.text assert "updates automatically while the job is active" in response.text @pytest.mark.asyncio async def test_job_cancel_page_renders_confirmation(self, app_client, seed_document_with_unlinked_job): _, client = app_client _, job_id = seed_document_with_unlinked_job response = client.get(f"/ui/jobs/{job_id}/cancel") assert response.status_code == 200 assert "Cancel Processing Job" in response.text assert "Cancel stops processing" in response.text assert "Cancel job" in response.text @pytest.mark.asyncio async def test_job_resubmit_page_renders_counts(self, app_client, seed_job): _, client = app_client job_id = await seed_job( filename="failed-resubmit.png", status=JobStatus.FAILED, transcription_text=None, error_detail="Timeout", ) response = client.get(f"/ui/jobs/{job_id}/resubmit") assert response.status_code == 200 assert "Resubmit Job" in response.text assert "Resubmittable Sources:" in response.text assert "Resubmit now" in response.text @pytest.mark.asyncio async def test_job_delete_page_blocks_deletion_when_processing(self, app_client): _, client = app_client async with session_scope() as session: doc = Document(name="Processing Doc") session.add(doc) await session.flush() job = Job(document_id=doc.id, status=JobStatus.PROCESSING) session.add(job) await session.commit() job_id = str(job.id) response = client.get(f"/ui/jobs/{job_id}/delete") assert response.status_code == 200 assert "Delete Processing Job" in response.text assert "Delete is blocked while the job is processing." in response.text @pytest.mark.asyncio async def test_job_delete_page_allows_deletion_for_queued_job(self, app_client, seed_document_with_unlinked_job): _, client = app_client _, job_id = seed_document_with_unlinked_job response = client.get(f"/ui/jobs/{job_id}/delete") assert response.status_code == 200 assert "Delete Processing Job" in response.text assert "Delete job and evidence" in response.text assert "immutable execution evidence" in response.text assert "Delete is blocked" not in response.text