generated from john/python-template
89 lines
3.5 KiB
Python
89 lines
3.5 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.models import Document
|
|
from transcription.models import Job
|
|
from transcription.services.documents import DocumentService
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.jobs import JobStatus
|
|
|
|
|
|
class TestJobService:
|
|
class TestBasicCRUD:
|
|
@pytest.mark.asyncio
|
|
async def test_create_job(self, job_service: JobService):
|
|
"""Test creating a job."""
|
|
|
|
def fake_job_factory():
|
|
return Job(document_id=uuid4())
|
|
|
|
await job_service.create_job(job=fake_job_factory())
|
|
|
|
async with job_service._session_scope() as session:
|
|
for _ in range(10):
|
|
await job_service.create_job(job=fake_job_factory(), session=session)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backpropagation(self, job_service: JobService, document_service: DocumentService):
|
|
"""Test that creating a job backpropagates to the related document."""
|
|
doc_id = uuid4()
|
|
document = Document(
|
|
id=doc_id,
|
|
filename="test.txt",
|
|
file_path="/path/to/test.txt",
|
|
)
|
|
await document_service.create_document(document=document)
|
|
job = Job(document_id=doc_id)
|
|
await job_service.create_job(job=job)
|
|
|
|
read_job = await job_service.read_job(job_id=job.id)
|
|
assert isinstance(read_job.document, Document)
|
|
assert read_job.document.id == document.id
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_reading_job(self, job_service: JobService):
|
|
"""Test reading a job."""
|
|
uuid = uuid4()
|
|
await job_service.create_job(job=Job(id=uuid, document_id=uuid4()))
|
|
job = await job_service.read_job(job_id=uuid)
|
|
assert job.id == uuid
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_updating_job(self, job_service: JobService):
|
|
"""Test updating a job."""
|
|
uuid = uuid4()
|
|
job = Job(id=uuid, document_id=uuid4())
|
|
async with job_service._session_scope() as session:
|
|
await job_service.create_job(job=job, session=session)
|
|
job.status = JobStatus.PROCESSING
|
|
await job_service.update_job(job=job, session=session)
|
|
read_job = await job_service.read_job(job_id=uuid, session=session)
|
|
assert read_job == job
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deleting_job(self, job_service: JobService):
|
|
"""Test deleting a job."""
|
|
|
|
class TestServiceMethods:
|
|
@pytest.mark.asyncio
|
|
async def test_query_jobs(self, job_service: JobService):
|
|
"""Test querying jobs."""
|
|
await job_service.create_job(job=Job(document_id=uuid4(), status=JobStatus.PROCESSING))
|
|
result = await job_service.query_jobs(status=JobStatus.PROCESSING)
|
|
jobs = {str(job.id).split("-")[0]: job.status for job in result}
|
|
assert len(jobs) == 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_jobs(self, job_service: JobService):
|
|
"""Test listing jobs."""
|
|
n = 5
|
|
for _ in range(n):
|
|
await job_service.create_job(job=Job(document_id=uuid4()))
|
|
jobs = await job_service.list_jobs()
|
|
assert len(jobs) == n
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_job_status(self, job_service: JobService):
|
|
"""Test marking a job with a new status."""
|