generated from john/python-template
92 lines
3.4 KiB
Python
92 lines
3.4 KiB
Python
"""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 = await 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.",
|
|
}
|
|
|
|
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)
|
|
|
|
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 = await 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)
|
|
|
|
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 |