diff --git a/src/transcription/db/runtime.py b/src/transcription/db/runtime.py index 703a70e..a0905f4 100644 --- a/src/transcription/db/runtime.py +++ b/src/transcription/db/runtime.py @@ -3,11 +3,13 @@ from collections.abc import AsyncGenerator from contextlib import asynccontextmanager from contextvars import ContextVar from dataclasses import dataclass +from functools import partial from sqlalchemy.ext.asyncio import AsyncEngine from sqlalchemy.ext.asyncio import async_sessionmaker from sqlalchemy.ext.asyncio import create_async_engine from sqlmodel.ext.asyncio.session import AsyncSession +from sqlmodel.pool import StaticPool from ..config import Settings from ..config import get_settings @@ -46,16 +48,21 @@ def _to_async_database_url(database_url: str) -> str: def _build_engine(settings: Settings) -> AsyncEngine: database_url = _to_async_database_url(settings.database_url) - connect_args: dict[str, object] = {} - if database_url.startswith("sqlite"): - connect_args["check_same_thread"] = False - return create_async_engine( + engine_factory = partial( + create_async_engine, url=database_url, echo=False, pool_pre_ping=True, - connect_args=connect_args, ) + if database_url.startswith("sqlite"): + sqlite_connect_settings = {"check_same_thread": settings.sqlite_check_same_thread} + engine_factory = partial(engine_factory, connect_args=sqlite_connect_settings) + if ":memory:" in database_url: + engine_factory = partial(engine_factory, poolclass=StaticPool) + + return engine_factory() + def initialize_database_runtime(*, settings: Settings | None = None) -> DatabaseRuntime: """Initialize lifespan-owned async DB resources once per process.""" diff --git a/src/transcription/services/base.py b/src/transcription/services/base.py index 307dff4..8ecf5d5 100644 --- a/src/transcription/services/base.py +++ b/src/transcription/services/base.py @@ -27,7 +27,7 @@ class ServiceBase(ABC): self.queue = queue or asyncio.Queue() @asynccontextmanager - async def _session_scope(self, session: AsyncSession | None): + async def _session_scope(self, session: AsyncSession | None = None): """Provide a transactional scope around a series of operations.""" if session is not None: # Reuse the provided session if one is passed in diff --git a/src/transcription/services/jobs.py b/src/transcription/services/jobs.py index aeb2b31..3c66e93 100644 --- a/src/transcription/services/jobs.py +++ b/src/transcription/services/jobs.py @@ -44,7 +44,7 @@ class JobService(ServiceBase): await _session.merge(job) await _session.commit() await _session.refresh(job) - return Job.model_copy(job) + return job async def delete_job(self, job: Job, session: AsyncSession | None = None) -> None: """Delete a job from the database.""" diff --git a/tests/conftest.py b/tests/conftest.py index 5a37231..1d6f4b0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -5,18 +5,33 @@ isolated, fast, and leave no artifacts on disk. """ import pytest -from sqlmodel import Session, SQLModel, create_engine +import pytest_asyncio +from sqlmodel import Session +from sqlmodel import SQLModel +from sqlmodel import create_engine from sqlmodel.pool import StaticPool +from transcription.db.runtime import dispose_database_runtime +from transcription.db.runtime import get_session + @pytest.fixture def session(): - """Provide a clean database session for each test.""" + """Provide a clean synchronous database session for sync tests.""" engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool, ) SQLModel.metadata.create_all(engine) - with Session(engine) as session: - yield session + with Session(engine) as sync_session: + yield sync_session + + +@pytest_asyncio.fixture +async def async_session(): + """Provide a clean asynchronous database session for async tests.""" + async with get_session() as async_session: + yield async_session + + await dispose_database_runtime() diff --git a/tests/services/test_job_service.py b/tests/services/test_job_service.py index df58d2f..7779c27 100644 --- a/tests/services/test_job_service.py +++ b/tests/services/test_job_service.py @@ -1,38 +1,75 @@ +from uuid import uuid4 + import pytest +from transcription.models import Job +from transcription.services.jobs import JobService +from transcription.services.jobs import JobStatus + + +@pytest.fixture +def job_service(): + """Provide a JobService instance for testing.""" + return JobService() + class TestJobService: class TestBasicCRUD: @pytest.mark.asyncio - async def test_create_job(self): + 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): + 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): + 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): + async def test_deleting_job(self, job_service: JobService): """Test deleting a job.""" class TestServiceMethods: @pytest.mark.asyncio - async def test_query_jobs(self): + 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): + async def test_list_jobs(self, job_service: JobService): """Test listing jobs.""" @pytest.mark.asyncio - async def test_mark_job_status(self): + 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): + async def test_multiple_operations(self, job_service: JobService): """Test multiple operations on jobs."""