From b607b0ccc4986250f1df726efd8789bea452f642 Mon Sep 17 00:00:00 2001 From: Adam Hernandez Date: Fri, 18 Sep 2026 20:00:50 -0700 Subject: [PATCH] test(utilities): isolate queue reader and writer transactions Use a file-backed SQLite fixture and commit setup before dispatch. The shared in-memory connection let cancellation readers roll back pending result writes, producing timing-dependent failures. Retain all existing outcome assertions and add a deterministic transaction-isolation regression. --- tests/utilities/test_job_queue.py | 56 +++++++++++++++++++++++++++++-- 1 file changed, 54 insertions(+), 2 deletions(-) diff --git a/tests/utilities/test_job_queue.py b/tests/utilities/test_job_queue.py index 8403042..ad30c2e 100644 --- a/tests/utilities/test_job_queue.py +++ b/tests/utilities/test_job_queue.py @@ -20,7 +20,9 @@ import pytest from sqlalchemy import select +from sqlalchemy.ext.asyncio import create_async_engine +from pullbox.models.base import Base from pullbox.models.config import SystemConfig from pullbox.models.issue import Issue, IssueStatus from pullbox.models.library import FileFormat, LibraryFile, LibraryRoot, MatchConfidence @@ -54,7 +56,37 @@ ) if TYPE_CHECKING: - from sqlalchemy.ext.asyncio import AsyncSession + from collections.abc import AsyncIterator + + from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + + +@pytest.fixture +async def async_engine(tmp_path: Path) -> AsyncIterator[AsyncEngine]: + """Use independent connections for concurrent dispatch and control readers.""" + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'queue.db'}") + async with engine.begin() as connection: + await connection.run_sync(Base.metadata.create_all) + try: + yield engine + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_queue_status_reader_preserves_pending_writer(session_factory) -> None: + """A control reader must not share or roll back the writer's transaction.""" + statement = select(SystemConfig.value).where(SystemConfig.key == "utility_worker_count") + async with session_factory() as writer: + writer.add(SystemConfig(key="utility_worker_count", value="2", value_type="int")) + await writer.flush() + async with session_factory() as reader: + observed = await reader.scalar(statement) + await writer.commit() + + async with session_factory() as reader: + assert await reader.scalar(statement) == "2" + assert observed is None # ── Test Executor (for batch execution tests) ────────────────── @@ -809,6 +841,7 @@ async def test_dispatch_completed_rollback_marks_parent_job_rolled_back( rollback_job = await mgr.queue_rollback_job(db_session, parent.id, created_by="admin") + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(parent) @@ -862,6 +895,7 @@ async def test_dispatch_failed_rollback_keeps_parent_completed( rollback_job = await mgr.queue_rollback_job(db_session, parent.id, created_by="admin") + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(parent) @@ -957,6 +991,7 @@ async def test_dispatch_updates_tracked_library_file_after_mass_convert( }, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1035,6 +1070,7 @@ async def test_dispatch_updates_issue_integrity_for_healthy_tracked_file( }, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1116,6 +1152,7 @@ async def _fake_search(series_id: int) -> dict[str, int]: }, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1197,9 +1234,11 @@ async def test_rollback_restores_quarantined_integrity_item_state( }, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) rollback_job = await mgr.queue_rollback_job(db_session, job.id, created_by="admin") + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1263,6 +1302,7 @@ async def test_dispatch_transitions_pausing_job_to_paused( config={}, ) + await db_session.commit() dispatch_task = asyncio.create_task(mgr.dispatch_next()) # Wait until job starts running for _ in range(50): @@ -1337,6 +1377,7 @@ async def test_dispatch_reuses_pending_items_for_resumed_jobs( ) await db_session.flush() + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1370,6 +1411,7 @@ async def test_dispatch_drains_into_next_queued_job_after_completion( config={"count": 1}, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(first) @@ -1541,6 +1583,7 @@ async def test_dispatch_drains_into_next_queued_job_after_pause( config={}, ) + await db_session.commit() dispatch_task = asyncio.create_task(mgr.dispatch_next()) for _ in range(50): await asyncio.sleep(0.05) @@ -1585,6 +1628,7 @@ async def test_dispatch_drains_into_next_queued_job_after_cancel( config={}, ) + await db_session.commit() dispatch_task = asyncio.create_task(mgr.dispatch_next()) for _ in range(50): await asyncio.sleep(0.05) @@ -1723,7 +1767,7 @@ async def test_recover_and_dispatch_starts_next_queued_job( created_at="2026-04-05T00:02:00+00:00", ) db_session.add_all([interrupted, queued]) - await db_session.flush() + await db_session.commit() recovered = await mgr.recover_and_dispatch() @@ -1750,6 +1794,7 @@ async def test_after_item_commit_receives_matching_item_payload( config={}, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1818,6 +1863,7 @@ def shutdown(self) -> None: config={"count": 5}, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -1894,6 +1940,7 @@ def shutdown(self) -> None: config={"count": 1}, ) + await db_session.commit() await mgr.dispatch_next() assert worker_counts == [4] @@ -1961,6 +2008,7 @@ def shutdown(self) -> None: config={"count": 3}, ) + await db_session.commit() dispatch_task = asyncio.create_task(mgr.dispatch_next()) try: # The event barrier proves incremental progress, not a subsecond startup SLA. @@ -2036,6 +2084,7 @@ def shutdown(self) -> None: config={"count": 2}, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -2075,6 +2124,7 @@ async def _boom(*args: Any, **kwargs: Any) -> Any: }, ) + await db_session.commit() await mgr.dispatch_next() await db_session.refresh(job) @@ -2106,6 +2156,7 @@ async def test_failed_item_promotes_guidance_logs_to_error( config={}, ) + await db_session.commit() await mgr.dispatch_next() result = await db_session.execute( @@ -2141,6 +2192,7 @@ async def test_utility_log_level_filters_db_history_and_file_output( config={}, ) + await db_session.commit() await mgr.dispatch_next() result = await db_session.execute(