From 9d5793f0ce3de8f915252f8a1a9a522d103c156e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 21 May 2026 16:01:59 -0400 Subject: [PATCH] fc3d: tick_due_sources Celery task on the scan queue Co-Authored-By: Claude Opus 4.7 (1M context) --- backend/app/tasks/scan.py | 73 +++++++++++++++++- tests/test_tick_due_sources.py | 132 +++++++++++++++++++++++++++++++++ 2 files changed, 201 insertions(+), 4 deletions(-) create mode 100644 tests/test_tick_due_sources.py diff --git a/backend/app/tasks/scan.py b/backend/app/tasks/scan.py index 54bbe56..8c7a400 100644 --- a/backend/app/tasks/scan.py +++ b/backend/app/tasks/scan.py @@ -1,16 +1,24 @@ -"""scan_directory task: walks the configured import path, creates an -ImportBatch, enumerates files into ImportTasks, and enqueues -import_media_file for each. +"""scan tasks: + +* ``scan_directory`` — walks the configured import path, creates an + ImportBatch, enumerates files into ImportTasks, and enqueues + import_media_file for each. +* ``tick_due_sources`` (FC-3d) — periodic Beat-fired scan that fires + ``download_source.delay()`` for sources due for a check. """ +import asyncio +from datetime import UTC, datetime from pathlib import Path from sqlalchemy import create_engine, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.orm import sessionmaker from ..celery_app import celery from ..config import get_config -from ..models import ImportBatch, ImportSettings, ImportTask +from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask +from ..services.scheduler_service import select_due_sources def _iter_import_files(import_root: Path): @@ -94,3 +102,60 @@ def scan_directory(self, triggered_by: str = "manual", backfill_phash.delay() return batch_id + + +# --- FC-3d: periodic source-check tick ------------------------------------ + + +def _async_session_factory(): + cfg = get_config() + engine = create_async_engine(cfg.database_url, future=True, pool_pre_ping=True) + return async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False), engine + + +async def _tick_due_sources_async() -> dict: + factory, engine = _async_session_factory() + try: + async with factory() as session: + due = await select_due_sources(session) + if not due: + return { + "due": 0, "queued": 0, "skipped_in_flight": 0, + "tick_at": datetime.now(UTC).isoformat(), + } + due_ids = [s.id for s in due] + in_flight_rows = (await session.execute( + select(DownloadEvent.source_id).where( + DownloadEvent.source_id.in_(due_ids), + DownloadEvent.status.in_(["pending", "running"]), + ) + )).scalars().all() + in_flight = set(in_flight_rows) + + # Local import: keep top of module light, avoid circular import + # at celery task discovery time. + from .download import download_source + + queued = 0 + for s in due: + if s.id in in_flight: + continue + session.add(DownloadEvent(source_id=s.id, status="pending")) + await session.commit() + download_source.delay(s.id) + queued += 1 + return { + "due": len(due), + "queued": queued, + "skipped_in_flight": len(in_flight), + "tick_at": datetime.now(UTC).isoformat(), + } + finally: + await engine.dispose() + + +@celery.task(name="backend.app.tasks.scan.tick_due_sources", bind=True) +def tick_due_sources(self) -> dict: + """FC-3d: scan for sources due for a periodic check and fire + ``download_source.delay()`` for each.""" + return asyncio.run(_tick_due_sources_async()) diff --git a/tests/test_tick_due_sources.py b/tests/test_tick_due_sources.py new file mode 100644 index 0000000..3df93e1 --- /dev/null +++ b/tests/test_tick_due_sources.py @@ -0,0 +1,132 @@ +"""FC-3d: tick_due_sources task tests. + +Verifies (a) due sources are queued, (b) sources with a pending/running +DownloadEvent are skipped, (c) the tick is read-only on Source itself +(no last_checked_at write), and (d) the result dict shape. +""" +import pytest +from sqlalchemy import select + +import backend.app.tasks.scan # noqa: F401 — side-effect: register celery task +from backend.app.celery_app import celery +from backend.app.models import Artist, DownloadEvent, Source + +pytestmark = pytest.mark.integration + + +class _FakeTask: + """Stand-in for the real Celery Task object — only `.delay()` matters.""" + + def __init__(self, sink: list): + self._sink = sink + + def delay(self, source_id: int): + self._sink.append(source_id) + + +@pytest.mark.asyncio +async def test_tick_returns_counts_dict_when_no_due_sources(db, monkeypatch): + """Conftest TRUNCATEs after every integration test, so we start clean.""" + delayed: list[int] = [] + monkeypatch.setattr( + "backend.app.tasks.download.download_source", _FakeTask(delayed), + ) + from backend.app.tasks.scan import _tick_due_sources_async + + result = await _tick_due_sources_async() + assert set(result.keys()) == {"due", "queued", "skipped_in_flight", "tick_at"} + assert result["due"] == 0 + assert result["queued"] == 0 + assert result["skipped_in_flight"] == 0 + assert delayed == [] + + +@pytest.mark.asyncio +async def test_tick_queues_due_source(db, monkeypatch): + artist = Artist(name="tick-q", slug="tick-q", auto_check=True) + db.add(artist) + await db.flush() + src = Source( + artist_id=artist.id, platform="patreon", url="https://tick-q", + enabled=True, consecutive_failures=0, + ) + db.add(src) + await db.commit() + + delayed: list[int] = [] + monkeypatch.setattr( + "backend.app.tasks.download.download_source", _FakeTask(delayed), + ) + from backend.app.tasks.scan import _tick_due_sources_async + result = await _tick_due_sources_async() + assert result["queued"] >= 1 + assert src.id in delayed + + pending_count = (await db.execute( + select(DownloadEvent.id).where( + DownloadEvent.source_id == src.id, + DownloadEvent.status == "pending", + ) + )).scalars().all() + assert len(pending_count) == 1 + + +@pytest.mark.asyncio +async def test_tick_skips_in_flight_source(db, monkeypatch): + artist = Artist(name="tick-inf", slug="tick-inf", auto_check=True) + db.add(artist) + await db.flush() + src = Source( + artist_id=artist.id, platform="patreon", url="https://tick-inf", + enabled=True, consecutive_failures=0, + ) + db.add(src) + await db.flush() + db.add(DownloadEvent(source_id=src.id, status="running")) + await db.commit() + + delayed: list[int] = [] + monkeypatch.setattr( + "backend.app.tasks.download.download_source", _FakeTask(delayed), + ) + from backend.app.tasks.scan import _tick_due_sources_async + result = await _tick_due_sources_async() + assert src.id not in delayed + assert result["skipped_in_flight"] >= 1 + + +@pytest.mark.asyncio +async def test_tick_does_not_modify_source_last_checked_at(db, monkeypatch): + """The tick creates DownloadEvent(pending) but never writes Source columns. + Source.last_checked_at is owned by DownloadService.finalize. + """ + artist = Artist(name="tick-ro", slug="tick-ro", auto_check=True) + db.add(artist) + await db.flush() + src = Source( + artist_id=artist.id, platform="patreon", url="https://tick-ro", + enabled=True, consecutive_failures=0, + ) + db.add(src) + await db.commit() + src_id = src.id + + before = (await db.execute( + select(Source.last_checked_at).where(Source.id == src_id) + )).scalar_one() + + monkeypatch.setattr( + "backend.app.tasks.download.download_source", _FakeTask([]), + ) + from backend.app.tasks.scan import _tick_due_sources_async + await _tick_due_sources_async() + + after = (await db.execute( + select(Source.last_checked_at).where(Source.id == src_id) + )).scalar_one() + assert before == after # still None / unchanged + + +def test_tick_task_is_registered(): + """Side-effect import at the top of this test file forces task module load.""" + assert "backend.app.tasks.scan.tick_due_sources" in celery.tasks