transcription/tests/test_worker.py

import asyncio
import logging
from typing import cast

import pytest

from transcription.errors import AppError
from transcription.errors import ErrorCategory
from transcription.services import ServiceBundle
from transcription.services.sources import SourceService
from transcription.worker import process_next_queued_job
from transcription.worker import run_worker_loop


@pytest.mark.asyncio
async def test_run_worker_loop_stops_on_non_retriable_exception(monkeypatch, caplog):
    """A programming error before a job is claimed stops the loop instead of spinning.

    Regression guard for review log [8]. This previously spun at the poll interval
    forever: the fault was classified non-retriable, logged, and then discarded, and
    it never reached the per-job retry machinery so nothing capped it. Measured at 20
    iterations in 1.2s before the fix.
    """
    calls = 0
    stop_event = asyncio.Event()

    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
        nonlocal calls
        _ = (session, session_factory, services)
        calls += 1
        raise RuntimeError("boom")

    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)

    with caplog.at_level(logging.CRITICAL):
        await asyncio.wait_for(
            run_worker_loop(stop_event=stop_event, poll_interval_seconds=0),
            timeout=5,
        )

    assert calls == 1
    assert "Worker loop stopped after a non-retriable error" in caplog.text


@pytest.mark.asyncio
async def test_run_worker_loop_survives_retriable_exception(monkeypatch, caplog):
    """A retriable fault is still suppressed so transient conditions do not stop work."""
    calls = 0
    stop_event = asyncio.Event()

    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
        nonlocal calls
        _ = (session, session_factory, services)
        calls += 1
        if calls == 1:
            raise AppError(
                "transient",
                category=ErrorCategory.EXTERNAL_PROVIDER,
                suggestion="retry",
                retriable=True,
            )
        stop_event.set()
        return False

    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)

    with caplog.at_level(logging.ERROR):
        await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0)

    assert calls == 2
    assert "Worker loop exception" in caplog.text


@pytest.mark.asyncio
async def test_run_worker_loop_reuses_one_bundle_across_jobs(monkeypatch):
    """HIGH-02: the provider client is built once per loop, not once per job."""
    stop_event = asyncio.Event()
    seen: list[object] = []
    closed = False

    class _Sources:
        async def aclose(self):
            nonlocal closed
            closed = True

    bundle = ServiceBundle(sources=cast("SourceService", _Sources()))
    monkeypatch.setattr(
        "transcription.worker.ServiceBundle.from_session_factory",
        classmethod(lambda _cls, _factory=None, **_kwargs: bundle),
    )

    async def _fake_process_next_queued_job(*, session=None, session_factory=None, services=None):
        _ = (session, session_factory)
        seen.append(services)
        if len(seen) >= 3:
            stop_event.set()
            return False
        return True

    monkeypatch.setattr("transcription.worker.process_next_queued_job", _fake_process_next_queued_job)

    await run_worker_loop(stop_event=stop_event, poll_interval_seconds=0)

    assert len(seen) == 3
    assert all(item is bundle for item in seen)
    assert closed is True


@pytest.mark.asyncio
async def test_process_next_closes_provider_for_the_bundle_it_owns(monkeypatch):
    closed = False

    class _Sources:
        async def aclose(self):
            nonlocal closed
            closed = True

    bundle = ServiceBundle(sources=cast("SourceService", _Sources()))
    monkeypatch.setattr(
        "transcription.worker.ServiceBundle.from_session_factory",
        classmethod(lambda _cls, _factory=None, **_kwargs: bundle),
    )

    async def _no_job(*, services, session):
        _ = (services, session)
        return False

    monkeypatch.setattr("transcription.worker.process_next_queued_job_workflow", _no_job)

    assert await process_next_queued_job() is False
    assert closed is True


@pytest.mark.asyncio
async def test_process_next_leaves_a_caller_owned_bundle_open(monkeypatch):
    """A bundle passed in belongs to the caller and must outlive one job."""
    closed = False

    class _Sources:
        async def aclose(self):
            nonlocal closed
            closed = True

    bundle = ServiceBundle(sources=cast("SourceService", _Sources()))

    async def _no_job(*, services, session):
        _ = (services, session)
        return False

    monkeypatch.setattr("transcription.worker.process_next_queued_job_workflow", _no_job)

    assert await process_next_queued_job(services=bundle) is False
    assert closed is False