bvandeusen
576e16d14d
Backfill events were STILL stranding empty after the timeout-ladder fix.
Worker logs showed the salvage path working ("Download timeout for
anduo/patreon after 1170.0s (18 files written)") but then:
Retry in 3s: DBAPIError(ConnectionDoesNotExistError: connection was
closed in the middle of operation)
...succeeded in 0.149s <- in-flight guard no-op
Root cause: DownloadService held the async + sync DB connections checked
out across the entire (≤19.5-min backfill) gallery-dl subprocess. The
server reaps the idle connection, so phase 3's first query hits a dead
socket. That DBAPIError trips download_source's autoretry_for, the retry
re-enters _phase1_setup, sees the event still 'running', returns
in_flight and no-ops — leaving the event to be stranded empty by the
recovery sweep. pool_pre_ping was already on both engines but can't help
a *held* connection (it only validates on pool checkout).
Fix:
- DownloadService.download_source closes the async + sync sessions after
phase 1, before the subprocess, so phase 3 re-acquires a live
connection (matches the class's "Phase 2 — no DB connection" docstring).
- The per-task async engine switches to NullPool so phase 3 always opens
a fresh connection rather than a pooled one the server may have reaped.
Tests: assert connections are released before gdl.download runs and the
event still finalizes; assert the task engine uses NullPool. Also fixes a
stale 1800s->1170s comment.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
694 lines
24 KiB
Python
694 lines
24 KiB
Python
"""DownloadService orchestrator tests.
|
|
|
|
GalleryDLService is mocked to return a canned DownloadResult; tests
|
|
provide on-disk files for the importer to attach_in_place.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from backend.app.models import Artist, DownloadEvent, ImportSettings, Source
|
|
from backend.app.services.credential_crypto import CredentialCrypto
|
|
from backend.app.services.credential_service import CredentialService
|
|
from backend.app.services.thumbnailer import Thumbnailer
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
@pytest.fixture
|
|
async def seed_artist_and_source(db, db_sync):
|
|
artist = Artist(name="Alice", slug="alice")
|
|
db.add(artist)
|
|
await db.flush()
|
|
source = Source(
|
|
artist_id=artist.id, platform="patreon",
|
|
url="https://patreon.com/alice", enabled=True,
|
|
config_overrides={},
|
|
)
|
|
db.add(source)
|
|
await db.commit()
|
|
return artist, source
|
|
|
|
|
|
def _make_jpg(path: Path, split: str = "h"):
|
|
"""Write a structured JPEG so phash isn't degenerate.
|
|
|
|
Solid-color images all phash-collapse (DCT of a flat image is zero).
|
|
Use `split='h'` (top/bottom halves) or `split='v'` (left/right) to
|
|
give each fixture file a distinct phash.
|
|
"""
|
|
from PIL import Image
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
im = Image.new("RGB", (64, 64), (255, 255, 255))
|
|
px = im.load()
|
|
if split == "h":
|
|
for y in range(32):
|
|
for x in range(64):
|
|
px[x, y] = (0, 0, 0)
|
|
else: # 'v'
|
|
for y in range(64):
|
|
for x in range(32):
|
|
px[x, y] = (0, 0, 0)
|
|
im.save(path, "JPEG")
|
|
|
|
|
|
def _make_fake_dl_result(
|
|
*, success=True, written_paths=None, quarantined_paths=None,
|
|
files_downloaded=0, error_type=None, error_message=None,
|
|
stdout="", stderr="",
|
|
):
|
|
return SimpleNamespace(
|
|
success=success,
|
|
url="https://patreon.com/alice",
|
|
artist_slug="alice",
|
|
platform="patreon",
|
|
files_downloaded=files_downloaded,
|
|
files_quarantined=len(quarantined_paths or []),
|
|
quarantined_paths=quarantined_paths or [],
|
|
written_paths=written_paths or [],
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
return_code=0 if success else 1,
|
|
error_type=error_type,
|
|
error_message=error_message,
|
|
duration_seconds=1.23,
|
|
started_at="2026-05-20T14:00:00+00:00",
|
|
completed_at="2026-05-20T14:01:00+00:00",
|
|
)
|
|
|
|
|
|
def _fake_gdl_with_result(result):
|
|
fake = MagicMock()
|
|
fake.download = AsyncMock(return_value=result)
|
|
fake._compute_run_stats = lambda *a, **k: {
|
|
"exit_code": 0, "downloaded_count": len(result.written_paths),
|
|
"skipped_count": 0, "per_item_failures": 0,
|
|
"warning_count": 0, "tier_gated_count": 0,
|
|
}
|
|
fake._extract_errors_warnings = lambda *a, **k: ""
|
|
fake._truncate_log = lambda x, **k: x
|
|
return fake
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_source_attaches_written_files(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
|
|
images_root = tmp_path / "images"
|
|
f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
|
|
f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
|
|
# Structured h/v split → distinct sha256 AND distinct phash;
|
|
# also pin phash_threshold=0 so even tiny perceptual similarity
|
|
# between the two doesn't trigger duplicate_phash.
|
|
_make_jpg(f1, split="h")
|
|
_make_jpg(f2, split="v")
|
|
|
|
fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
|
|
success=True,
|
|
written_paths=[str(f1), str(f2)],
|
|
files_downloaded=2,
|
|
stdout=f"{f1}\n{f2}\n",
|
|
))
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
sync_settings.phash_threshold = 0 # only exact phash collisions dedup
|
|
|
|
importer = Importer(
|
|
session=db_sync, images_root=images_root,
|
|
import_root=images_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root),
|
|
settings=sync_settings,
|
|
)
|
|
crypto = CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
|
|
cred_service = CredentialService(db, crypto)
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
event_id = await svc.download_source(source.id)
|
|
|
|
ev = (await db.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
assert ev.status == "ok"
|
|
assert ev.files_count == 2
|
|
assert ev.metadata_["import_summary"]["attached"] == 2
|
|
assert ev.metadata_["run_stats"]["downloaded_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_source_skipped_when_disabled(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
source.enabled = False
|
|
await db.commit()
|
|
|
|
fake_gdl = MagicMock()
|
|
fake_gdl.download = AsyncMock()
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=tmp_path,
|
|
import_root=tmp_path,
|
|
thumbnailer=Thumbnailer(images_root=tmp_path),
|
|
settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, MagicMock())
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
event_id = await svc.download_source(source.id)
|
|
|
|
ev = (await db.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
assert ev.status == "skipped"
|
|
assert ev.error == "source disabled"
|
|
fake_gdl.download.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_in_flight_idempotency_returns_existing_event(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
from backend.app.services.download_service import DownloadService
|
|
|
|
_artist, source = seed_artist_and_source
|
|
running = DownloadEvent(source_id=source.id, status="running")
|
|
db.add(running)
|
|
await db.commit()
|
|
await db.refresh(running)
|
|
|
|
fake_gdl = MagicMock()
|
|
fake_gdl.download = AsyncMock()
|
|
importer_stub = MagicMock()
|
|
cred_service = CredentialService(db, MagicMock())
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer_stub, cred_service=cred_service,
|
|
)
|
|
event_id = await svc.download_source(source.id)
|
|
assert event_id == running.id
|
|
fake_gdl.download.assert_not_called()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patreon_retry_caches_campaign_id(
|
|
db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
|
|
):
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
|
|
failed = _make_fake_dl_result(
|
|
success=False, written_paths=[], stdout="",
|
|
stderr="[patreon][error] Failed to extract campaign ID for alice\n",
|
|
)
|
|
succeeded = _make_fake_dl_result(success=True, written_paths=[])
|
|
|
|
download_calls = []
|
|
|
|
async def fake_download(url, **k):
|
|
download_calls.append(url)
|
|
return failed if len(download_calls) == 1 else succeeded
|
|
|
|
fake_gdl = MagicMock()
|
|
fake_gdl.download = fake_download
|
|
fake_gdl._compute_run_stats = lambda *a, **k: {
|
|
"exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
|
|
"per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
|
|
}
|
|
fake_gdl._extract_errors_warnings = lambda *a, **k: ""
|
|
fake_gdl._truncate_log = lambda x, **k: x
|
|
|
|
monkeypatch.setattr(
|
|
"backend.app.services.download_service.resolve_campaign_id",
|
|
AsyncMock(return_value="99"),
|
|
)
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=tmp_path,
|
|
import_root=tmp_path,
|
|
thumbnailer=Thumbnailer(images_root=tmp_path),
|
|
settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, MagicMock())
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
await svc.download_source(source.id)
|
|
|
|
assert len(download_calls) == 2
|
|
assert "id:99" in download_calls[1]
|
|
|
|
overrides = db_sync.execute(
|
|
select(Source.config_overrides).where(Source.id == source.id)
|
|
).scalar_one()
|
|
assert overrides.get("patreon_campaign_id") == "99"
|
|
|
|
|
|
# --- FC-3d: finalize hook updates Source health columns -------------------
|
|
|
|
|
|
async def _seed_source_with_health(db, *, failures=0, last_error=None, suffix="fz"):
|
|
artist = Artist(name=f"alice-{suffix}", slug=f"alice-{suffix}")
|
|
db.add(artist)
|
|
await db.flush()
|
|
source = Source(
|
|
artist_id=artist.id, platform="patreon",
|
|
url=f"https://patreon.com/alice-{suffix}", enabled=True,
|
|
consecutive_failures=failures, last_error=last_error,
|
|
)
|
|
db.add(source)
|
|
await db.flush()
|
|
event = DownloadEvent(source_id=source.id, status="running")
|
|
db.add(event)
|
|
await db.commit()
|
|
return source.id, event.id
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finalize_ok_resets_failures_and_clears_error(db):
|
|
from backend.app.services.download_service import DownloadService
|
|
|
|
source_id, event_id = await _seed_source_with_health(
|
|
db, failures=3, last_error="prev", suffix="ok",
|
|
)
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=None,
|
|
gdl=None, importer=None, cred_service=None,
|
|
)
|
|
await svc._finalize_event_and_source(
|
|
event_id=event_id, source_id=source_id, status="ok",
|
|
error_message=None,
|
|
)
|
|
|
|
row = (await db.execute(
|
|
select(
|
|
Source.consecutive_failures, Source.last_error, Source.last_checked_at,
|
|
).where(Source.id == source_id)
|
|
)).one()
|
|
assert row.consecutive_failures == 0
|
|
assert row.last_error is None
|
|
assert row.last_checked_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finalize_error_increments_failures_and_sets_error(db):
|
|
from backend.app.services.download_service import DownloadService
|
|
|
|
source_id, event_id = await _seed_source_with_health(
|
|
db, failures=2, last_error=None, suffix="err",
|
|
)
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=None,
|
|
gdl=None, importer=None, cred_service=None,
|
|
)
|
|
await svc._finalize_event_and_source(
|
|
event_id=event_id, source_id=source_id, status="error",
|
|
error_message="boom",
|
|
)
|
|
|
|
row = (await db.execute(
|
|
select(
|
|
Source.consecutive_failures, Source.last_error, Source.last_checked_at,
|
|
).where(Source.id == source_id)
|
|
)).one()
|
|
assert row.consecutive_failures == 3
|
|
assert row.last_error == "boom"
|
|
assert row.last_checked_at is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_finalize_skipped_preserves_failures_clears_error(db):
|
|
from backend.app.services.download_service import DownloadService
|
|
|
|
source_id, event_id = await _seed_source_with_health(
|
|
db, failures=4, last_error="stale", suffix="skip",
|
|
)
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=None,
|
|
gdl=None, importer=None, cred_service=None,
|
|
)
|
|
await svc._finalize_event_and_source(
|
|
event_id=event_id, source_id=source_id, status="skipped",
|
|
error_message=None,
|
|
)
|
|
|
|
row = (await db.execute(
|
|
select(
|
|
Source.consecutive_failures, Source.last_error, Source.last_checked_at,
|
|
).where(Source.id == source_id)
|
|
)).one()
|
|
assert row.consecutive_failures == 4
|
|
assert row.last_error is None
|
|
assert row.last_checked_at is not None
|
|
|
|
|
|
# --- Plan #544: backfill lifecycle + PARTIAL → status=ok -------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_decrements_after_run(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
"""When backfill_runs_remaining > 0 going in, a non-clean / non-empty
|
|
run decrements by 1 — operator gets N runs to complete the deep scan
|
|
before tick mode resumes."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
source.backfill_runs_remaining = 3
|
|
await db.commit()
|
|
|
|
images_root = tmp_path / "images"
|
|
f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
|
|
_make_jpg(f1, split="h")
|
|
|
|
fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
|
|
success=True, written_paths=[str(f1)], files_downloaded=1,
|
|
stdout=f"{f1}\n",
|
|
))
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=images_root, import_root=images_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
await svc.download_source(source.id)
|
|
|
|
remaining = (await db.execute(
|
|
select(Source.backfill_runs_remaining).where(Source.id == source.id)
|
|
)).scalar_one()
|
|
assert remaining == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_auto_resets_on_clean_zero_files(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
"""A clean run (rc=0) that downloaded zero files means the backfill
|
|
queue drained — reset to 0 immediately instead of wasting the rest of
|
|
the N-run budget on no-op walks."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
source.backfill_runs_remaining = 3
|
|
await db.commit()
|
|
|
|
fake_result = _make_fake_dl_result(
|
|
success=True, written_paths=[], files_downloaded=0,
|
|
)
|
|
fake_gdl = _fake_gdl_with_result(fake_result)
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=tmp_path / "images",
|
|
import_root=tmp_path / "images",
|
|
thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
|
|
settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
await svc.download_source(source.id)
|
|
|
|
remaining = (await db.execute(
|
|
select(Source.backfill_runs_remaining).where(Source.id == source.id)
|
|
)).scalar_one()
|
|
assert remaining == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_mode_does_not_touch_backfill_counter(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
"""When backfill_runs_remaining is already 0, downloads don't go
|
|
negative or otherwise mutate the counter."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
assert source.backfill_runs_remaining == 0
|
|
|
|
fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
|
|
success=True, written_paths=[], files_downloaded=0,
|
|
))
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=tmp_path / "images",
|
|
import_root=tmp_path / "images",
|
|
thumbnailer=Thumbnailer(images_root=tmp_path / "images"),
|
|
settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
await svc.download_source(source.id)
|
|
|
|
remaining = (await db.execute(
|
|
select(Source.backfill_runs_remaining).where(Source.id == source.id)
|
|
)).scalar_one()
|
|
assert remaining == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_partial_error_type_maps_to_ok_status(
|
|
db, db_sync, tmp_path, seed_artist_and_source,
|
|
):
|
|
"""A non-zero gallery-dl exit accompanied by PARTIAL error_type
|
|
(real files were downloaded before the run was cut short) is treated
|
|
as status=ok by the orchestrator — next tick continues."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.gallery_dl import ErrorType
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
|
|
images_root = tmp_path / "images"
|
|
f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
|
|
_make_jpg(f1, split="h")
|
|
|
|
fake_result = _make_fake_dl_result(
|
|
success=False, written_paths=[str(f1)], files_downloaded=1,
|
|
error_type=ErrorType.PARTIAL,
|
|
error_message="Downloaded 1 file; run did not complete in budget",
|
|
stdout=f"{f1}\n",
|
|
)
|
|
fake_gdl = _fake_gdl_with_result(fake_result)
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=images_root, import_root=images_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
event_id = await svc.download_source(source.id)
|
|
|
|
ev = (await db.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
assert ev.status == "ok"
|
|
assert ev.error is None
|
|
# And the source's failure counter isn't bumped — PARTIAL isn't a failure.
|
|
src_after = (await db.execute(
|
|
select(Source.consecutive_failures).where(Source.id == source.id)
|
|
)).scalar_one()
|
|
assert src_after == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_download_enqueues_thumbnail_and_ml_per_attached_image(
|
|
db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
|
|
):
|
|
"""Operator-flagged 2026-06-01: downloaded images stayed at
|
|
thumbnail_path=NULL until periodic backfill swept them up, surfacing
|
|
as broken-thumbnail tiles in the gallery. Importer.attach_in_place
|
|
deliberately skips inline generation; the calling code MUST enqueue
|
|
the thumbnail + ML tasks per attached image (matching the pattern in
|
|
tasks/import_file.py)."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
|
|
images_root = tmp_path / "images"
|
|
f1 = images_root / "alice" / "patreon" / "post" / "a.jpg"
|
|
f2 = images_root / "alice" / "patreon" / "post" / "b.jpg"
|
|
_make_jpg(f1, split="h")
|
|
_make_jpg(f2, split="v")
|
|
|
|
fake_gdl = _fake_gdl_with_result(_make_fake_dl_result(
|
|
success=True, written_paths=[str(f1), str(f2)],
|
|
files_downloaded=2, stdout=f"{f1}\n{f2}\n",
|
|
))
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
sync_settings.phash_threshold = 0
|
|
importer = Importer(
|
|
session=db_sync, images_root=images_root, import_root=images_root,
|
|
thumbnailer=Thumbnailer(images_root=images_root), settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True))
|
|
|
|
# Capture the IDs that the orchestrator hands off to each Celery task.
|
|
# The .delay() shim runs inside DownloadService._phase3_persist (lazy
|
|
# imports under ..tasks.thumbnail / ..tasks.ml), so monkeypatch the
|
|
# symbols on those modules.
|
|
thumb_calls: list[int] = []
|
|
ml_calls: list[int] = []
|
|
from backend.app.tasks import ml as ml_mod
|
|
from backend.app.tasks import thumbnail as thumb_mod
|
|
monkeypatch.setattr(
|
|
thumb_mod.generate_thumbnail, "delay",
|
|
lambda image_id: thumb_calls.append(image_id),
|
|
)
|
|
monkeypatch.setattr(
|
|
ml_mod.tag_and_embed, "delay",
|
|
lambda image_id: ml_calls.append(image_id),
|
|
)
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
await svc.download_source(source.id)
|
|
|
|
# Two files attached → two thumbnail enqueues + two ML enqueues, IDs
|
|
# match the actually-imported records (not ad-hoc — drawn from the
|
|
# importer's returned image_id so future supersede paths stay covered).
|
|
assert len(thumb_calls) == 2
|
|
assert len(ml_calls) == 2
|
|
assert sorted(thumb_calls) == sorted(ml_calls)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_releases_db_connections_before_subprocess(
|
|
db, db_sync, tmp_path, seed_artist_and_source, monkeypatch,
|
|
):
|
|
"""Phase 1's DB connections must be released BEFORE the (up to ~19.5-min
|
|
in backfill) gallery-dl subprocess, so they don't idle-die and strand
|
|
phase 3 with ConnectionDoesNotExistError (Anduo #40014). Spy on the
|
|
async session close and assert it happened before gdl.download runs;
|
|
phase 3 must still finalize the event."""
|
|
from backend.app.services.download_service import DownloadService
|
|
from backend.app.services.importer import Importer
|
|
|
|
_artist, source = seed_artist_and_source
|
|
|
|
closed = {"async": False}
|
|
orig_close = db.close
|
|
|
|
async def spy_close():
|
|
closed["async"] = True
|
|
await orig_close()
|
|
|
|
monkeypatch.setattr(db, "close", spy_close)
|
|
|
|
seen = {}
|
|
|
|
async def fake_download(url, **k):
|
|
seen["closed_before_download"] = closed["async"]
|
|
return _make_fake_dl_result(success=True, written_paths=[], stdout="")
|
|
|
|
fake_gdl = MagicMock()
|
|
fake_gdl.download = fake_download
|
|
fake_gdl._compute_run_stats = lambda *a, **k: {
|
|
"exit_code": 0, "downloaded_count": 0, "skipped_count": 0,
|
|
"per_item_failures": 0, "warning_count": 0, "tier_gated_count": 0,
|
|
}
|
|
fake_gdl._extract_errors_warnings = lambda *a, **k: ""
|
|
fake_gdl._truncate_log = lambda x, **k: x
|
|
|
|
sync_settings = db_sync.execute(
|
|
select(ImportSettings).where(ImportSettings.id == 1)
|
|
).scalar_one()
|
|
importer = Importer(
|
|
session=db_sync, images_root=tmp_path, import_root=tmp_path,
|
|
thumbnailer=Thumbnailer(images_root=tmp_path), settings=sync_settings,
|
|
)
|
|
cred_service = CredentialService(
|
|
db, CredentialCrypto(tmp_path / "key.b64", bootstrap_ok=True)
|
|
)
|
|
|
|
svc = DownloadService(
|
|
async_session=db, sync_session=db_sync,
|
|
gdl=fake_gdl, importer=importer, cred_service=cred_service,
|
|
)
|
|
event_id = await svc.download_source(source.id)
|
|
|
|
assert seen["closed_before_download"] is True
|
|
ev = (await db.execute(
|
|
select(DownloadEvent).where(DownloadEvent.id == event_id)
|
|
)).scalar_one()
|
|
assert ev.status == "ok"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_task_engine_uses_nullpool():
|
|
"""The per-task async engine must use NullPool so phase 3 re-acquires a
|
|
fresh connection instead of a pooled one the server reaped during the
|
|
long subprocess (Anduo #40014)."""
|
|
from sqlalchemy.pool import NullPool
|
|
|
|
from backend.app.tasks._async_session import async_session_factory
|
|
|
|
_factory, engine = async_session_factory()
|
|
try:
|
|
assert isinstance(engine.sync_engine.pool, NullPool)
|
|
finally:
|
|
await engine.dispose()
|