started test_job_service

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] = {}
+    engine_factory = partial(
-    if database_url.startswith("sqlite"):
+        create_async_engine,
-        connect_args["check_same_thread"] = False
-    return 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."""

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

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."""

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:
+    with Session(engine) as sync_session:
-        yield 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()

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."""