Files
FabledCurator/tests/test_external_worker.py
bvandeusen 19b962f1a7
CI / lint (push) Successful in 3s
CI / frontend-build (push) Successful in 20s
CI / backend-lint-and-test (push) Successful in 30s
CI / integration (push) Successful in 3m31s
feat(b3): ml-worker becomes optional — embed-only role, decoupled GPU coordination, cpu-embed switch
The ml-worker's ONLY processing role is now the CPU whole-image embed fallback
(tag_and_embed renamed embed_image — Camie tagging was retired #1189 and the
name kept implying otherwise; videos were already handled agent-style: frame
sampling + mean-pool). Detection/cropping/CCIP stay GPU-agent-only, and their
completion is judged per-pipeline: ccip by gpu_job rows, siglip by concept
regions at the current model version — never by image_record.siglip_embedding.
A CPU embed therefore can NEVER close crop work for the agent (regression test
pins this; only the whole-image 'embed' job, the same artifact, is satisfied).

Making removal actually safe (operator will drop the container):
- GPU-queue coordination (enqueue_gpu_backfill, recover_orphaned_gpu_jobs,
  reprocess_gpu_jobs) moved verbatim to tasks/gpu_queue.py on the maintenance
  quick lane — it lived on the 'ml' queue only by module colocation, which made
  the ml-worker a hard dependency of the whole agent pipeline.
- New ml_settings.cpu_embed_enabled (migration 0074, default ON so agent-less
  installs keep working): OFF stops the four import hooks queueing embed work
  nothing will consume and no-ops the manual backfill; switch lives on the
  renamed 'CPU embedding backfill' card.
- NB heads training / auto-apply still run on the ml image (sklearn) — a stack
  that removes the container gives those up too.

Deploy note: in-flight messages under the old task names are dropped by the
new workers; the 60s orphan sweep + hourly backfill re-fire under the new
names immediately.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
2026-07-02 16:53:08 -04:00

254 lines
8.7 KiB
Python

"""Integration tests for the external-link download worker (tasks/external)."""
import io
import zipfile
import pytest
from PIL import Image
from sqlalchemy import func, select, update
import backend.app.tasks.external as ext
from backend.app.models import (
Artist,
ExternalLink,
ImageProvenance,
ImageRecord,
Post,
PostAttachment,
Source,
)
from backend.app.services.external_fetch import FetchResult
pytestmark = pytest.mark.integration
def _structured_jpeg_bytes():
# A non-uniform image so it survives the import filters (a solid color would
# phash-collapse / fail thresholds).
im = Image.new("RGB", (256, 256))
px = im.load()
for y in range(256):
for x in range(256):
px[x, y] = (x, y, (x + y) % 256)
buf = io.BytesIO()
im.save(buf, "JPEG")
return buf.getvalue()
class _FakeLock:
def acquire(self, blocking=False):
return True
def release(self):
pass
class _FakeRedis:
def lock(self, name, timeout=None, blocking=False):
return _FakeLock()
def _seed(db_sync, *, host="pixeldrain", status="pending"):
# Key per host so a single test can seed several (artist name is UNIQUE).
artist = Artist(name=f"Ext {host}", slug=f"ext-{host}")
db_sync.add(artist)
db_sync.flush()
source = Source(
artist_id=artist.id, platform="patreon",
url=f"https://patreon.com/{host}", enabled=True, config_overrides={},
)
db_sync.add(source)
db_sync.flush()
post = Post(
source_id=source.id, artist_id=artist.id, external_post_id=f"EXT-{host}",
post_url=f"https://patreon.com/posts/{host}",
)
db_sync.add(post)
db_sync.flush()
link = ExternalLink(
post_id=post.id, artist_id=artist.id, host=host,
url=f"https://{host}.test/file", status=status,
)
db_sync.add(link)
db_sync.commit()
return post, link
def test_fetch_external_link_downloads_and_attaches(db_sync, tmp_path, monkeypatch):
post, link = _seed(db_sync)
link_id = link.id
monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
def fake_fetch(host, url, dest_dir, **kwargs):
dest_dir.mkdir(parents=True, exist_ok=True)
f = dest_dir / "film.bin" # non-art → PostAttachment (no thumb/ML enqueue)
f.write_bytes(b"a film pack")
return FetchResult(files=[f], bytes=f.stat().st_size)
monkeypatch.setattr(ext, "fetch_external", fake_fetch)
out = ext.fetch_external_link(link_id)
assert out.get("files") == 1
db_sync.expire_all()
refreshed = db_sync.get(ExternalLink, link_id)
assert refreshed.status == "downloaded"
assert refreshed.completed_at is not None
# The file was captured as a PostAttachment linked to the SAME post.
atts = db_sync.execute(
select(PostAttachment).where(PostAttachment.post_id == post.id)
).scalars().all()
assert len(atts) == 1
def test_refetch_same_link_keeps_canonical_file(db_sync, tmp_path, monkeypatch):
"""#859: re-fetching the SAME link (identical bytes into the same per-link
staging dir) must NOT unlink the canonical file out from under its record.
The first fetch imports the image in place; the second dedups on sha256, and
the path-safe unlink recognises the file IS the record's backing file and
keeps it — so the post never 404s on playback."""
_, link = _seed(db_sync, host="pixeldrain")
link_id = link.id
monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
def fake_fetch(host, url, dest_dir, **kwargs):
dest_dir.mkdir(parents=True, exist_ok=True)
f = dest_dir / "clip.jpg" # art → ImageRecord (imported in place)
f.write_bytes(_structured_jpeg_bytes())
return FetchResult(files=[f], bytes=f.stat().st_size)
monkeypatch.setattr(ext, "fetch_external", fake_fetch)
import backend.app.tasks.ml as ml_mod
import backend.app.tasks.thumbnail as thumb_mod
monkeypatch.setattr(ml_mod.embed_image, "delay", lambda i: None)
monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: None)
out = ext.fetch_external_link(link_id)
assert out["images"] == 1
db_sync.expire_all()
rec = db_sync.execute(select(ImageRecord)).scalars().one()
canonical = tmp_path / rec.path # absolute path collapses the join
assert canonical.exists()
# Re-arm the SAME link and fetch again — same staging dir, identical bytes.
db_sync.execute(
update(ExternalLink).where(ExternalLink.id == link_id)
.values(status="pending", attempts=0, completed_at=None)
)
db_sync.commit()
out2 = ext.fetch_external_link(link_id)
assert out2["images"] == 0 # deduped, no new record
db_sync.expire_all()
# Exactly one record still, and its file survived the dedup unlink (the bug).
assert db_sync.execute(select(func.count(ImageRecord.id))).scalar() == 1
assert canonical.exists()
def test_fetch_external_link_records_failure(db_sync, tmp_path, monkeypatch):
_, link = _seed(db_sync)
link_id = link.id
monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
monkeypatch.setattr(
ext, "fetch_external",
lambda *a, **k: FetchResult(error="host 503"),
)
ext.fetch_external_link(link_id)
db_sync.expire_all()
refreshed = db_sync.get(ExternalLink, link_id)
assert refreshed.status == "failed"
assert refreshed.attempts == 1
assert "503" in refreshed.last_error
def test_fetch_external_link_dead_letters_at_threshold(db_sync, tmp_path, monkeypatch):
_, link = _seed(db_sync)
link.attempts = ext.DEAD_LETTER_THRESHOLD - 1
db_sync.commit()
link_id = link.id
monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
monkeypatch.setattr(ext, "fetch_external", lambda *a, **k: FetchResult(error="nope"))
ext.fetch_external_link(link_id)
db_sync.expire_all()
assert db_sync.get(ExternalLink, link_id).status == "dead"
def test_fetch_external_link_skips_non_claimable(db_sync, tmp_path, monkeypatch):
_, link = _seed(db_sync, status="downloaded")
link_id = link.id
called = []
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
monkeypatch.setattr(
ext, "fetch_external",
lambda *a, **k: called.append(1) or FetchResult(),
)
out = ext.fetch_external_link(link_id)
assert out.get("skipped") == "not claimable"
assert called == [] # never fetched
def test_sweep_enqueues_pending_and_retryable(db_sync, monkeypatch):
_, l1 = _seed(db_sync, host="pixeldrain")
_, l2 = _seed(db_sync, host="mega", status="failed")
# A dead one must NOT be swept.
_, l3 = _seed(db_sync, host="dropbox", status="dead")
enqueued = []
monkeypatch.setattr(ext.fetch_external_link, "delay", lambda lid: enqueued.append(lid))
out = ext.sweep_external_links()
assert out["enqueued"] == 2
assert set(enqueued) == {l1.id, l2.id}
def test_downloaded_archive_gets_provenance_and_tagging(db_sync, tmp_path, monkeypatch):
"""An archive downloaded by the worker is extracted, provenance-linked to the
post, AND queued for tagging + thumbnails — exactly like an extracted zip on
the normal download path."""
post, link = _seed(db_sync, host="mega")
link_id = link.id
monkeypatch.setattr(ext, "IMAGES_ROOT", tmp_path)
monkeypatch.setattr(ext, "_redis", lambda: _FakeRedis())
def fake_fetch(host, url, dest_dir, **kwargs):
dest_dir.mkdir(parents=True, exist_ok=True)
zpath = dest_dir / "pack.zip"
with zipfile.ZipFile(zpath, "w") as z:
z.writestr("art.jpg", _structured_jpeg_bytes())
return FetchResult(files=[zpath], bytes=zpath.stat().st_size)
monkeypatch.setattr(ext, "fetch_external", fake_fetch)
tagged, thumbed = [], []
import backend.app.tasks.ml as ml_mod
import backend.app.tasks.thumbnail as thumb_mod
monkeypatch.setattr(ml_mod.embed_image, "delay", lambda i: tagged.append(i))
monkeypatch.setattr(thumb_mod.generate_thumbnail, "delay", lambda i: thumbed.append(i))
out = ext.fetch_external_link(link_id)
assert out["images"] >= 1
db_sync.expire_all()
assert db_sync.get(ExternalLink, link_id).status == "downloaded"
# Extracted member is provenance-linked to the SAME post (as extracted zips).
prov = db_sync.execute(
select(ImageProvenance).where(ImageProvenance.post_id == post.id)
).scalars().all()
assert len(prov) >= 1
member_ids = {p.image_record_id for p in prov}
# Tagging + thumbnails queued for exactly those member images.
assert set(tagged) == member_ids
assert set(thumbed) == member_ids