Skip to content

test: isolate utility queue transactions in CI #155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
DeusExTaco merged 1 commit into from
Sep 19, 2026
+54 −2
Merged

test: isolate utility queue transactions in CI #155

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 54 additions & 2 deletions tests/utilities/test_job_queue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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) ──────────────────
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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()

Expand All @@ -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)
Expand Down Expand Up @@ -1818,6 +1863,7 @@ def shutdown(self) -> None:
config={"count": 5},
)

await db_session.commit()
await mgr.dispatch_next()

await db_session.refresh(job)
Expand Down Expand Up @@ -1894,6 +1940,7 @@ def shutdown(self) -> None:
config={"count": 1},
)

await db_session.commit()
await mgr.dispatch_next()

assert worker_counts == [4]
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -2036,6 +2084,7 @@ def shutdown(self) -> None:
config={"count": 2},
)

await db_session.commit()
await mgr.dispatch_next()

await db_session.refresh(job)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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(
Expand Down
Loading