generated from john/python-template
79 lines
3.0 KiB
Python
79 lines
3.0 KiB
Python
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from transcription.config import Settings
|
|
from transcription.db.runtime import get_session_factory
|
|
from transcription.models import Job
|
|
from transcription.services.jobs import JobService
|
|
from transcription.services.jobs import JobStatus
|
|
|
|
|
|
@pytest.fixture
|
|
def job_service(default_settings: Settings) -> JobService:
|
|
"""Provide a JobService instance for testing."""
|
|
session_factory = get_session_factory(settings=default_settings)
|
|
return JobService(session_factory=session_factory)
|
|
|
|
|
|
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_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."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_job_status(self, job_service: JobService):
|
|
"""Test marking a job with a new status."""
|
|
|
|
class TestMultipleOperations:
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_operations(self, job_service: JobService):
|
|
"""Test multiple operations on jobs."""
|