80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
|
"""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
|