"""Tests that a failed ingestion job actually gets resubmitted to arq. arq only auto-retries a job on its own `Retry` exception (or CancelledError) — never on an arbitrary exception a task raises. Before this, app/services/ingestion/runner.py:_run_ingestion just recorded IngestionJob.status = "retrying" in the database and re-raised the original exception, which arq treated as a terminal failure: the DB said "retrying" forever, but nothing ever put the job back on the queue. """ from __future__ import annotations import uuid import pytest from arq import Retry from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import session_scope from app.repositories.provider_repository import ( create_ingestion_job, get_ingestion_job, get_or_create_provider, ) from app.services.ingestion.runner import _run_ingestion class _BoomError(Exception): pass async def _make_job(*, attempts_already_used: int = 0) -> uuid.UUID: async with session_scope() as session: provider = await get_or_create_provider(session, "tmdb", "TMDB") job = await create_ingestion_job( session, provider_id=provider.id, job_type="movie_sync", idempotency_key=f"test:{uuid.uuid4()}", entity_type="movie", external_id="603", ) job.attempts = attempts_already_used await session.flush() return job.id @pytest.mark.asyncio async def test_run_ingestion_raises_arq_retry_while_attempts_remain(db_session: AsyncSession): job_id = await _make_job(attempts_already_used=0) async def failing_work(session, provider): raise _BoomError("simulated failure") with pytest.raises(Retry): await _run_ingestion(job_id, "test_ingestion", failing_work) async with session_scope() as session: job = await get_ingestion_job(session, job_id) assert job.status.value == "retrying" assert job.attempts == 1 @pytest.mark.asyncio async def test_run_ingestion_raises_original_error_once_attempts_exhausted( db_session: AsyncSession, ): job_id = await _make_job(attempts_already_used=4) # default max_attempts is 5 async def failing_work(session, provider): raise _BoomError("simulated failure") with pytest.raises(_BoomError): await _run_ingestion(job_id, "test_ingestion", failing_work) async with session_scope() as session: job = await get_ingestion_job(session, job_id) assert job.status.value == "failed" assert job.attempts == 5