Revamped the Documents, People, & Jobs too.

tests/ui/test_sources_actions.py

@@ -0,0 +1,98 @@
+"""Action handler tests for Source CRUD mutations."""
+
+from pathlib import Path
+
+import pytest
+from sqlmodel import select
+
+from transcription.db import session_scope
+from transcription.db.models import Document, Job, JobSource, Source
+
+
+@pytest.mark.integration
+class TestSourcesActionHandlers:
+    """Verify POST/mutation routes for Source revisions and deletions."""
+
+    @pytest.mark.asyncio
+    async def test_upsert_revision_for_source_success(self, app_client, seed_job):
+        _, client = app_client
+        job_id = seed_job(
+            filename="revision-source.png",
+            transcription_text="automated raw transcription 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)
+
+        payload = {
+            "revised_text": "Curated human transcription text by editor.",
+        }
+
+        response = client.post(f"/ui/sources/{source_id}", data=payload, follow_redirects=True)
+
+        assert response.status_code == 200
+        assert "Revision saved" in response.text or "Curated human transcription text by editor." in response.text
+
+        async with session_scope() as session:
+            updated_source = await session.get(Source, source_id)
+            assert updated_source is not None
+            assert updated_source.revised_text == "Curated human transcription text by editor."
+            assert updated_source.date_revised is not None
+
+    @pytest.mark.asyncio
+    async def test_delete_unlinked_source_success(self, app_client):
+        _, client = app_client
+
+        async with session_scope() as session:
+            doc = Document(name="Unlinked Source Doc", document_type="memo")
+            session.add(doc)
+            await session.flush()
+
+            source = Source(
+                document_id=doc.id,
+                page_number=1,
+                upload_name="orphan_page.png",
+                filename="orphan_page.png",
+                file_path="/tmp/orphan_page.png",
+            )
+            session.add(source)
+            await session.commit()
+            source_id = str(source.id)
+
+        response = client.post(f"/ui/sources/{source_id}/delete", follow_redirects=True)
+
+        assert response.status_code == 200
+        assert "Source deleted" in response.text or "Archival Source Media" in response.text
+
+        async with session_scope() as session:
+            deleted_source = await session.get(Source, source_id)
+            assert deleted_source is None
+
+    @pytest.mark.asyncio
+    async def test_delete_source_blocked_when_job_linked(self, app_client, seed_job):
+        _, client = app_client
+        job_id = seed_job(filename="job-linked-source.png", transcription_text="job 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.post(f"/ui/sources/{source_id}/delete", follow_redirects=True)
+
+        assert response.status_code == 200
+        assert "Delete is only available for unlinked sources." in response.text or "linked" in response.text.lower()
+
+        async with session_scope() as session:
+            source_still_exists = await session.get(Source, source_id)
+            assert source_still_exists is not None