bvandeusenandClaude Opus 5
970d17f98a
CI and images / lint (push) Successful in 3s
CI and images / extension-version (push) Successful in 3s
CI and images / frontend-build (push) Successful in 21s
CI and images / backend-lint-and-test (push) Successful in 32s
CI and images / integration (push) Successful in 2m13s
CI and images / sign-extension (push) Successful in 2s
CI and images / build-agent (push) Successful in 5s
CI and images / build-web (push) Successful in 1m45s
CI and images / smoke-web (push) Successful in 1m6s
CI and images / promote (push) Skipped
Run 7422 integration: two #830 tests died with `_seed_seen() got an unexpected keyword argument 'post_id'`. Not those tests' fault — I defined a second `_seed_seen` at the BOTTOM of the module, taking a list of media instead of one key, and Python's last definition wins for every call site in the file including the ones 500 lines above it. Renamed to `_seed_all_seen` and made it loop over the existing one-key helper, so there is one definition of what seeding the ledger means. The nearby failure mode is worth naming: a helper defined below the tests that use it is invisible at the point of use, and shadowing produces a TypeError in code nobody touched. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR
1428 lines
57 KiB
Python
1428 lines
57 KiB
Python
"""PatreonIngester tests (native ingester build step 3).
|
|
|
|
The HTTP client and the media downloader are stubbed (injected seams), so these
|
|
exercise the ingester's own logic — mode behavior, the two-tier skip, cursor
|
|
emission, the budget cutoff, and the Postgres seen-ledger — without network or a
|
|
real CDN. The ledger is real (a sync sessionmaker bound to the test engine), so
|
|
the tier-1 skip and the idempotent mark-seen run against actual rows.
|
|
"""
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
import pytest
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from backend.app.models import (
|
|
Artist,
|
|
PatreonFailedMedia,
|
|
PatreonSeenMedia,
|
|
Source,
|
|
)
|
|
from backend.app.services.gallery_dl import ErrorType
|
|
from backend.app.services.ingest_core import _CANARY_MIN_SAMPLE
|
|
from backend.app.services.patreon_client import (
|
|
MediaItem,
|
|
PatreonAPIError,
|
|
PatreonAuthError,
|
|
PatreonDriftError,
|
|
)
|
|
from backend.app.services.patreon_downloader import MediaOutcome, PostRecordOutcome
|
|
from backend.app.services.patreon_ingester import PatreonIngester, _ledger_key
|
|
|
|
pytestmark = pytest.mark.integration
|
|
|
|
|
|
# --- fakes ----------------------------------------------------------------
|
|
|
|
|
|
def _media(post_id, n, *, filehash=None, kind="images"):
|
|
# Deterministic 32-hex-ish key, unique per (post_id, n).
|
|
fh = filehash if filehash is not None else f"{post_id}{n}".encode().hex().ljust(32, "0")[:32]
|
|
return MediaItem(
|
|
url=f"https://cdn.patreon.com/{fh}/{n}.jpg",
|
|
filename=f"{n}.jpg",
|
|
kind=kind,
|
|
filehash=fh,
|
|
post_id=post_id,
|
|
)
|
|
|
|
|
|
class _FakeClient:
|
|
"""Stub PatreonClient. `pages` is a list of (page_cursor, [posts]); each post
|
|
is (post_id, [MediaItem]). `raise_on_first` lets a test trip drift."""
|
|
|
|
def __init__(self, pages, raise_exc=None, empty_body=False, gated=None,
|
|
published=None):
|
|
self._pages = pages
|
|
self._raise_exc = raise_exc
|
|
# {post_id: ISO-8601 published_at} for the revisit window. Absent → the
|
|
# date reads as None, which is how EVERY test written before the window
|
|
# existed keeps its old behaviour: an unreadable date is never inside
|
|
# the horizon, so the count early-out stands alone.
|
|
self._published = dict(published or {})
|
|
# empty_body simulates a body-field schema break: every post comes back
|
|
# with no content (the #862 canary's trip condition).
|
|
self._empty_body = empty_body
|
|
# #874: post_ids the account can't view → current_user_can_view=False,
|
|
# so the walk must skip them entirely (no media, no post-record).
|
|
self._gated = set(gated or ())
|
|
self.consumed_posts = 0
|
|
|
|
def iter_posts(self, campaign_id, cursor=None):
|
|
if self._raise_exc is not None:
|
|
raise self._raise_exc
|
|
for page_cursor, posts in self._pages:
|
|
for post_id, media in posts:
|
|
self.consumed_posts += 1
|
|
# Carry feed attributes (title/post_type/content) like the real
|
|
# client — ingest_core reads these for the per-post diagnostic.
|
|
content = "" if self._empty_body else f"<p>body {post_id}</p>"
|
|
yield (
|
|
{
|
|
"id": post_id,
|
|
"_media": media,
|
|
"attributes": {
|
|
"title": f"Post {post_id}",
|
|
"post_type": "image_file",
|
|
"content": content,
|
|
"current_user_can_view": post_id not in self._gated,
|
|
},
|
|
},
|
|
{},
|
|
page_cursor,
|
|
)
|
|
|
|
def extract_media(self, post, included_index):
|
|
return post["_media"]
|
|
|
|
def post_meta(self, post):
|
|
return {
|
|
"title": post.get("id"),
|
|
"date": self._published.get(str(post.get("id") or "")),
|
|
}
|
|
|
|
@staticmethod
|
|
def post_is_gated(post):
|
|
return (post.get("attributes") or {}).get("current_user_can_view") is False
|
|
|
|
@staticmethod
|
|
def post_record_key(post):
|
|
pid = str(post.get("id") or "")
|
|
return (f"post:{pid}", pid) if pid else None
|
|
|
|
|
|
class _FakeDownloader:
|
|
"""Stub PatreonDownloader. Honors the injected is_seen (tier-1); media in
|
|
`on_disk` report skipped_disk (tier-2); media in `quarantine` report
|
|
quarantined; everything else downloads."""
|
|
|
|
def __init__(self, tmp_path, on_disk=None, quarantine=None, error=None):
|
|
self.tmp_path = tmp_path
|
|
self.on_disk = set(on_disk or ())
|
|
self.quarantine = set(quarantine or ())
|
|
self.error = set(error or ())
|
|
self.download_calls = 0
|
|
self.post_records = 0
|
|
self.post_revisits = 0
|
|
|
|
def download_post(self, post, media_items, artist_slug, *, is_seen,
|
|
should_stop=lambda: False, recapture=False):
|
|
outcomes = []
|
|
for m in media_items:
|
|
if should_stop():
|
|
break
|
|
seen = is_seen(m)
|
|
# Recapture mirrors the real downloader: a seen item isn't tier-1
|
|
# short-circuited — it falls through so an on-disk file surfaces as
|
|
# skipped_disk (for source_filehash backfill); a seen file NOT on
|
|
# disk is left alone (skipped_seen, never refetched).
|
|
if seen and not recapture:
|
|
outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
|
|
elif _ledger_key(m) in self.on_disk:
|
|
p = self.tmp_path / f"{m.post_id}_{m.filename}"
|
|
outcomes.append(MediaOutcome(media=m, status="skipped_disk", path=p, error=None))
|
|
elif seen:
|
|
outcomes.append(MediaOutcome(media=m, status="skipped_seen", path=None, error=None))
|
|
elif _ledger_key(m) in self.error:
|
|
self.download_calls += 1
|
|
outcomes.append(MediaOutcome(media=m, status="error", path=None, error="boom"))
|
|
elif _ledger_key(m) in self.quarantine:
|
|
self.download_calls += 1
|
|
q = self.tmp_path / "_quarantine" / f"{m.post_id}_{m.filename}"
|
|
outcomes.append(MediaOutcome(media=m, status="quarantined", path=q, error="bad image"))
|
|
else:
|
|
self.download_calls += 1
|
|
p = self.tmp_path / f"{m.post_id}_{m.filename}"
|
|
p.write_bytes(b"x")
|
|
outcomes.append(MediaOutcome(media=m, status="downloaded", path=p, error=None))
|
|
return outcomes
|
|
|
|
def write_post_record(self, post, artist_slug, *, revisit=False):
|
|
self.post_records += 1
|
|
if revisit:
|
|
self.post_revisits += 1
|
|
attrs_ = post.get("attributes") or {}
|
|
body_ = attrs_.get("content")
|
|
# Mirrors the real downloaders' revisit contract: a re-read whose body
|
|
# came back empty writes NOTHING rather than blanking a stored one.
|
|
if revisit and not (isinstance(body_, str) and body_.strip()):
|
|
return PostRecordOutcome(
|
|
path=None, post_type=attrs_.get("post_type"),
|
|
title=attrs_.get("title"), body_chars=0,
|
|
)
|
|
p = self.tmp_path / f"{post.get('id')}__post.json"
|
|
p.write_text("{}")
|
|
attrs = post.get("attributes") or {}
|
|
body = attrs.get("content")
|
|
return PostRecordOutcome(
|
|
path=p,
|
|
post_type=attrs.get("post_type"),
|
|
title=attrs.get("title"),
|
|
body_chars=len(body) if isinstance(body, str) else 0,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
async def source_id(db):
|
|
artist = Artist(name="Ingest", slug="ingest")
|
|
db.add(artist)
|
|
await db.flush()
|
|
source = Source(
|
|
artist_id=artist.id, platform="patreon",
|
|
url="https://patreon.com/ingest", enabled=True, config_overrides={},
|
|
)
|
|
db.add(source)
|
|
await db.commit()
|
|
return source.id
|
|
|
|
|
|
def _ingester(sync_engine, tmp_path, client, downloader):
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
return PatreonIngester(
|
|
images_root=tmp_path, cookies_path=None, session_factory=factory,
|
|
client=client, downloader=downloader,
|
|
)
|
|
|
|
|
|
def _count_ledger(sync_engine, source_id):
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
return s.execute(
|
|
select(func.count(PatreonSeenMedia.id)).where(
|
|
PatreonSeenMedia.source_id == source_id
|
|
)
|
|
).scalar_one()
|
|
|
|
|
|
# --- tick -----------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_downloads_unseen_and_marks_seen(source_id, sync_engine, tmp_path):
|
|
m1, m2 = _media("p1", 1), _media("p1", 2)
|
|
client = _FakeClient([(None, [("p1", [m1, m2])])])
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.return_code == 0
|
|
assert result.files_downloaded == 2
|
|
assert len(result.written_paths) == 2
|
|
# plan #704: structured run_stats carry the real counts.
|
|
assert result.run_stats["downloaded_count"] == 2
|
|
assert result.posts_processed == 1
|
|
# 2 media keys + 1 synthetic post key (body/links recaptured per post).
|
|
assert _count_ledger(sync_engine, source_id) == 3
|
|
# The post body + links are captured for media posts too (rides the walk).
|
|
assert len(result.post_record_paths) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_quarantined_media_surfaced_in_result(source_id, sync_engine, tmp_path):
|
|
"""plan #704 (#4): a media that fails validation is reported as quarantined —
|
|
a real files_quarantined count + paths + run_stats — not a blank 0, and is
|
|
NOT written or marked seen."""
|
|
m1, m2 = _media("p1", 1), _media("p1", 2)
|
|
client = _FakeClient([(None, [("p1", [m1, m2])])])
|
|
downloader = _FakeDownloader(tmp_path, quarantine={_ledger_key(m2)})
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
assert result.files_downloaded == 1 # only m1
|
|
assert result.files_quarantined == 1 # m2
|
|
assert len(result.quarantined_paths) == 1
|
|
assert result.run_stats["quarantined_count"] == 1
|
|
assert result.run_stats["downloaded_count"] == 1
|
|
assert len(result.written_paths) == 1 # quarantined NOT written
|
|
# Quarantined media is NOT marked seen (a fixed file may be re-fetched);
|
|
# m1 + the synthetic post key (body/links captured per post) = 2.
|
|
assert _count_ledger(sync_engine, source_id) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_skips_seen_via_ledger(source_id, sync_engine, tmp_path):
|
|
m1 = _media("p1", 1)
|
|
# Pre-seed the ledger with m1's key.
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
|
|
s.commit()
|
|
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
assert result.files_downloaded == 0
|
|
assert downloader.download_calls == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_early_out_after_threshold(source_id, sync_engine, tmp_path):
|
|
# Three posts, one already-seen media each; threshold 2 → stop before the 3rd.
|
|
seen_media = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
for m in seen_media:
|
|
s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m), post_id=m.post_id))
|
|
s.commit()
|
|
|
|
pages = [(None, [(m.post_id, [m]) for m in seen_media])]
|
|
client = _FakeClient(pages)
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
|
|
)
|
|
assert result.success is True
|
|
# Stopped after the 2nd contiguous seen item — never consumed the 3rd post.
|
|
assert client.consumed_posts == 2
|
|
|
|
|
|
# --- backfill -------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_reaches_bottom_signals_complete(source_id, sync_engine, tmp_path):
|
|
pages = [
|
|
(None, [("p1", [_media("p1", 1)])]),
|
|
("CUR2", [("p2", [_media("p2", 1)])]),
|
|
]
|
|
client = _FakeClient(pages)
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
# rc 0 + error_type None is what _apply_backfill_lifecycle reads as "complete".
|
|
assert result.return_code == 0
|
|
assert result.error_type is None
|
|
assert result.files_downloaded == 2
|
|
# The page-2 cursor is carried structurally for checkpointing (plan #704).
|
|
assert result.cursor == "CUR2"
|
|
assert result.posts_processed == 2
|
|
assert result.run_stats["downloaded_count"] == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_checkpoints_cursor_mid_walk(source_id, sync_engine, tmp_path, db):
|
|
"""plan #705 #6: the cursor is persisted to the DB at each page boundary
|
|
DURING the walk (not just at the end via phase 3), so a worker SIGKILL
|
|
mid-chunk resumes near the crash. After the walk the source carries the last
|
|
page's cursor — written by the ingester, not phase 3 (which the unit run
|
|
never reaches)."""
|
|
pages = [
|
|
(None, [("p1", [_media("p1", 1)])]),
|
|
("CUR2", [("p2", [_media("p2", 1)])]),
|
|
("CUR3", [("p3", [_media("p3", 1)])]),
|
|
]
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
co = (await db.execute(
|
|
select(Source.config_overrides).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
assert co.get("_backfill_cursor") == "CUR3"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_persists_live_posts_count(source_id, sync_engine, tmp_path, db):
|
|
"""plan #704 #5: the ingester writes a monotonic _backfill_posts absolute
|
|
DURING the walk (seeded from prior chunks via posts_base), EXCLUDING the
|
|
re-walked resume page so the count doesn't inflate across chunks."""
|
|
# Resume from CUR2 (a prior chunk left off here): its page is re-walked and
|
|
# must NOT be re-counted. posts_base=4 stands in for the prior chunks.
|
|
pages = [
|
|
("CUR2", [("p1", [_media("p1", 1)])]), # the resumed page — not counted
|
|
("CUR3", [("p2", [_media("p2", 1)])]), # net-new
|
|
("CUR4", [("p3", [_media("p3", 1)])]), # net-new
|
|
]
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), _FakeDownloader(tmp_path))
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
resume_cursor="CUR2", posts_base=4,
|
|
)
|
|
co = (await db.execute(
|
|
select(Source.config_overrides).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
# 4 prior + 2 net-new (p2, p3); p1 on the resumed CUR2 page is excluded.
|
|
assert co.get("_backfill_posts") == 6
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_does_not_persist_posts(source_id, sync_engine, tmp_path, db):
|
|
"""A tick has no resumable backfill state, so it must not write _backfill_posts
|
|
(the badge is a backfill/recovery concept)."""
|
|
ing = _ingester(
|
|
sync_engine, tmp_path,
|
|
_FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]),
|
|
_FakeDownloader(tmp_path),
|
|
)
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
co = (await db.execute(
|
|
select(Source.config_overrides).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
assert "_backfill_posts" not in (co or {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_does_not_checkpoint_cursor(source_id, sync_engine, tmp_path, db):
|
|
"""A tick has no resumable backfill state, so it must not write a cursor."""
|
|
ing = _ingester(
|
|
sync_engine, tmp_path,
|
|
_FakeClient([("CUR2", [("p1", [_media("p1", 1)])])]),
|
|
_FakeDownloader(tmp_path),
|
|
)
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
co = (await db.execute(
|
|
select(Source.config_overrides).where(Source.id == source_id)
|
|
)).scalar_one()
|
|
assert "_backfill_cursor" not in (co or {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_budget_cut_returns_partial_with_progress(
|
|
source_id, sync_engine, tmp_path, monkeypatch,
|
|
):
|
|
# Deterministic clock: start=0; post1 gate=10 (ok); post1's per-media
|
|
# should_stop=20 (still ok, so post1's item downloads); post1 live-progress
|
|
# read=200; post2 gate=250 (>budget → cut). The per-media should_stop check
|
|
# (added 2026-06-07 to bound media-dense posts) reads the clock once more per
|
|
# post, so the sequence carries the extra tick. run() lives in ingest_core
|
|
# now, so patch the clock there.
|
|
import backend.app.services.ingest_core as core
|
|
ticks = iter([0.0, 10.0, 20.0, 200.0, 250.0])
|
|
last = [0.0]
|
|
|
|
def fake_monotonic():
|
|
try:
|
|
last[0] = next(ticks)
|
|
except StopIteration:
|
|
pass
|
|
return last[0]
|
|
|
|
monkeypatch.setattr(core.time, "monotonic", fake_monotonic)
|
|
|
|
pages = [
|
|
("CUR1", [("p1", [_media("p1", 1)])]),
|
|
("CUR2", [("p2", [_media("p2", 1)])]),
|
|
]
|
|
client = _FakeClient(pages)
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
resume_cursor=None, time_budget_seconds=100.0,
|
|
)
|
|
assert result.error_type == ErrorType.PARTIAL
|
|
assert result.return_code == -1
|
|
assert result.files_downloaded == 1 # post1 downloaded before the cut
|
|
assert downloader.download_calls == 1 # post2's body never ran
|
|
# The checkpoint is the page we were CUT ON (CUR2 — entered, cursor emitted,
|
|
# then the budget check broke before processing it); the next chunk resumes
|
|
# there. Structured (plan #704), matching the old parse_last_cursor(last).
|
|
assert result.cursor == "CUR2"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_write_live_progress_updates_running_event(
|
|
source_id, sync_engine, tmp_path, db,
|
|
):
|
|
"""plan #709: live counts land in the RUNNING event's metadata.live; a
|
|
finalized event is left alone (status guard)."""
|
|
from backend.app.models import DownloadEvent
|
|
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path))
|
|
running = DownloadEvent(source_id=source_id, status="running")
|
|
done = DownloadEvent(source_id=source_id, status="ok")
|
|
db.add_all([running, done])
|
|
await db.commit()
|
|
|
|
counts = {"downloaded": 3, "skipped": 1, "errors": 0, "quarantined": 0, "posts": 4}
|
|
ing._write_live_progress(running.id, counts)
|
|
ing._write_live_progress(done.id, {"downloaded": 9}) # guarded: status != running
|
|
|
|
live = (await db.execute(
|
|
select(DownloadEvent.metadata_).where(DownloadEvent.id == running.id)
|
|
)).scalar_one()
|
|
assert live["live"] == counts
|
|
done_meta = (await db.execute(
|
|
select(DownloadEvent.metadata_).where(DownloadEvent.id == done.id)
|
|
)).scalar_one()
|
|
assert "live" not in (done_meta or {})
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_still_running_reads_backfill_state(source_id, sync_engine, tmp_path, db):
|
|
"""plan #708 B4: _still_running reflects the source's `_backfill_state` — the
|
|
flag an operator Stop pops to cancel a live chunk."""
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient([]), _FakeDownloader(tmp_path))
|
|
assert ing._still_running(source_id) is False # absent → not running
|
|
src = (await db.execute(select(Source).where(Source.id == source_id))).scalar_one()
|
|
src.config_overrides = {"_backfill_state": "running"}
|
|
await db.commit()
|
|
assert ing._still_running(source_id) is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_stops_when_operator_cancels_mid_walk(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""plan #708 B4: with the running state latched, a later page boundary that
|
|
finds it gone (Stop) bails the chunk with PARTIAL instead of finishing."""
|
|
pages = [
|
|
("CUR2", [("p1", [_media("p1", 1)])]),
|
|
("CUR3", [("p2", [_media("p2", 1)])]),
|
|
("CUR4", [("p3", [_media("p3", 1)])]),
|
|
]
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient(pages), downloader)
|
|
|
|
# Latch on the page-2 boundary (running), then "Stop" before page 3.
|
|
calls = {"n": 0}
|
|
|
|
def fake_still_running(_source_id):
|
|
calls["n"] += 1
|
|
return calls["n"] == 1
|
|
|
|
ing._still_running = fake_still_running
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert result.error_type == ErrorType.PARTIAL
|
|
assert "Stopped by operator" in result.error_message
|
|
assert downloader.download_calls == 1 # only p1 ran; cancelled before p2
|
|
|
|
|
|
# --- recovery -------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_bypasses_seen_ledger(source_id, sync_engine, tmp_path):
|
|
m1 = _media("p1", 1)
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
s.add(PatreonSeenMedia(source_id=source_id, filehash=_ledger_key(m1), post_id="p1"))
|
|
s.commit()
|
|
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recovery",
|
|
)
|
|
# Ledger says seen, but recovery bypasses tier-1 → re-downloaded for a fresh
|
|
# pHash evaluation under the current threshold.
|
|
assert result.files_downloaded == 1
|
|
assert downloader.download_calls == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_tier2_disk_still_skips(source_id, sync_engine, tmp_path):
|
|
m1 = _media("p1", 1)
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
# File still on disk (a kept image) → tier-2 spares it even under recovery.
|
|
downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recovery",
|
|
)
|
|
assert result.files_downloaded == 0
|
|
assert downloader.download_calls == 0
|
|
# Disk-skip reconciles the media key + the synthetic post key (recovery
|
|
# recaptures the body/links per post) = 2.
|
|
assert _count_ledger(sync_engine, source_id) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_recaptures_body_for_already_downloaded_post(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Phase 5 guarantee: a post whose media is ALREADY on disk still gets its
|
|
body + external links recaptured on a re-walk. Re-downloading existing media
|
|
can never fill links the system never had — so the recapture rides the walk
|
|
itself, and a normal backfill is the backfill."""
|
|
m1 = _media("p1", 1)
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert result.files_downloaded == 0 # media already on disk
|
|
assert len(result.post_record_paths) == 1 # body/links recaptured anyway
|
|
assert downloader.post_records == 1
|
|
|
|
|
|
def _seed_seen(sync_engine, source_id, key, post_id=None):
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
s.add(PatreonSeenMedia(source_id=source_id, filehash=key, post_id=post_id))
|
|
s.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_backfill_skips_already_captured_post_but_recapture_forces_it(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""#830: a plain backfill GATES post-record capture on the seen-ledger, so a
|
|
post whose `post:<id>` key is already seen is NOT recaptured (the operator's
|
|
'Industrial Lust body missing' bug). RECAPTURE mode bypasses that gate and
|
|
re-captures the body/links even though the post key is seen — without
|
|
re-downloading the on-disk media."""
|
|
m1 = _media("p1", 1)
|
|
# Pre-seed BOTH the media key and the synthetic post key as already seen.
|
|
_seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1")
|
|
_seed_seen(sync_engine, source_id, "post:p1", post_id="p1")
|
|
|
|
# 1) Plain backfill: post key is seen → gate skips body recapture.
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
res_backfill = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert res_backfill.post_record_paths == [] # gated → not recaptured
|
|
assert downloader.post_records == 0
|
|
|
|
# 2) Recapture: gate bypassed → body recaptured; media NOT re-downloaded;
|
|
# the on-disk file is surfaced for source_filehash relink.
|
|
client2 = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader2 = _FakeDownloader(tmp_path, on_disk={_ledger_key(m1)})
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
|
|
res_recap = ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recapture",
|
|
)
|
|
assert len(res_recap.post_record_paths) == 1 # gate bypassed → recaptured
|
|
assert downloader2.post_records == 1
|
|
assert res_recap.files_downloaded == 0 # no media re-download
|
|
assert downloader2.download_calls == 0
|
|
# The on-disk media is surfaced as a (path, CDN url, post_id) relink triple
|
|
# (post_id drives the #1288 image→post back-link in phase 3).
|
|
assert len(res_recap.relink_source_paths) == 1
|
|
rel_path, rel_url, rel_post_id = res_recap.relink_source_paths[0]
|
|
assert rel_url == m1.url
|
|
assert rel_post_id == m1.post_id
|
|
# Diagnostic summary surfaces the post-record + relink counts (operator reads
|
|
# this off the event stdout to see what a recapture actually did).
|
|
assert "1 post-record(s), 1 relinked" in res_recap.stdout
|
|
# #842: per-post handling is written into the SAME run stdout (reuses the
|
|
# existing "Raw stdout" panel) — post_type + body length, so an empty body is
|
|
# self-explanatory by its post_type.
|
|
assert f"post p1 [image_file] body: {len('<p>body p1</p>')} chars" in res_recap.stdout
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gated_post_skipped_entirely_no_media_no_record(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""#874: a tier-gated post (current_user_can_view=False) is skipped ENTIRELY
|
|
— no media download AND no post-record stub — while an accessible post in the
|
|
same walk is fully ingested. Patreon serves only blurred locked-preview media
|
|
for gated posts; downloading it produced unusable images."""
|
|
gm = _media("gated", 1)
|
|
am = _media("open", 1)
|
|
client = _FakeClient([(None, [("gated", [gm]), ("open", [am])])], gated={"gated"})
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
|
|
# Only the accessible post's media downloaded; the gated post contributed
|
|
# nothing — no file, no post-record.
|
|
assert result.files_downloaded == 1
|
|
# (substring check on the basename only — the tmp dir name contains "gated")
|
|
assert not any(p.endswith("gated_1.jpg") for p in result.written_paths)
|
|
assert downloader.download_calls == 1
|
|
assert len(result.post_record_paths) == 1 # only the open post
|
|
assert downloader.post_records == 1
|
|
# The gated post left NO trace in the seen-ledger (no media key, no post key):
|
|
# only the open post's media key + synthetic post key are recorded.
|
|
assert _count_ledger(sync_engine, source_id) == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gated_posts_are_reported_in_run_stats(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""A gated post must be COUNTED, not just skipped.
|
|
|
|
`make_run_stats` has always declared `tier_gated_count` and
|
|
DownloadDetailModal has always rendered it, but the native path never
|
|
populated it — so a walk that skipped N paywalled posts reported 0, and a
|
|
creator we've lost access to looked exactly like a creator who stopped
|
|
posting. Asserted on the run_stats key the UI actually reads (not the
|
|
ingester's internal counter) so the guard tracks the property, not a name.
|
|
"""
|
|
gated = [(f"g{i}", [_media(f"g{i}", 1)]) for i in range(3)]
|
|
am = _media("open", 1)
|
|
client = _FakeClient(
|
|
[(None, [*gated, ("open", [am])])],
|
|
gated={"g0", "g1", "g2"},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
|
|
assert result.run_stats["tier_gated_count"] == 3
|
|
# Not conflated with the other skip tiers: seen/on-disk skips are a
|
|
# different fact from "we aren't allowed to see this".
|
|
assert result.run_stats["skipped_count"] == 0
|
|
assert result.run_stats["downloaded_count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fully_gated_walk_classifies_tier_limited_but_still_succeeds(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""A walk where everything was paywalled is a SUCCESS with a reason.
|
|
|
|
Before this, it returned `error_type=None` — identical to a creator who
|
|
simply hadn't posted. success/return_code are asserted deliberately: they
|
|
are what keep the walk "completed" for the backfill lifecycle and status
|
|
"ok" for source health, so a source we can't see never accrues
|
|
consecutive_failures or a backoff it hasn't earned.
|
|
"""
|
|
gated = [(f"g{i}", [_media(f"g{i}", 1)]) for i in range(2)]
|
|
client = _FakeClient([(None, gated)], gated={"g0", "g1"})
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
|
|
assert result.error_type == ErrorType.TIER_LIMITED
|
|
assert result.success is True
|
|
assert result.return_code == 0
|
|
assert "2 posts" in result.error_message
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tier_limited_fires_even_when_some_media_downloaded(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Gated posts classify TIER_LIMITED even on a walk that DID download.
|
|
|
|
Mirrors gallery-dl's long-standing rule (`_categorize_error`), which does
|
|
not condition on a zero download count either. The fact worth surfacing is
|
|
"there is content here you aren't paying for", and that is equally true in a
|
|
week we also got the cheaper posts. Pinned because the obvious-looking
|
|
stricter rule (`downloaded == 0`) would silently diverge the two backends.
|
|
"""
|
|
client = _FakeClient(
|
|
[(None, [("g0", [_media("g0", 1)]), ("open", [_media("open", 1)])])],
|
|
gated={"g0"},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
|
|
assert result.files_downloaded == 1
|
|
assert result.error_type == ErrorType.TIER_LIMITED
|
|
assert result.success is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_run_stats_reports_zero_gated_when_nothing_is_gated(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The counter reports a real zero on a clean walk — so a 0 in the UI means
|
|
'nothing was gated', not 'this path never counted'."""
|
|
client = _FakeClient([(None, [("p1", [_media("p1", 1)])])])
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
|
|
assert result.run_stats["tier_gated_count"] == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recapture_does_not_refetch_seen_media_missing_from_disk(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Recapture must NOT re-download a seen item that's absent from disk (that's
|
|
recovery's job) — it returns skipped_seen and contributes no relink pair."""
|
|
m1 = _media("p1", 1)
|
|
_seed_seen(sync_engine, source_id, _ledger_key(m1), post_id="p1")
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
# NOT in on_disk → seen + missing.
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recapture",
|
|
)
|
|
assert result.files_downloaded == 0
|
|
assert downloader.download_calls == 0
|
|
assert result.relink_source_paths == []
|
|
|
|
|
|
# --- dead-letter ledger (plan #705 #7) ------------------------------------
|
|
|
|
|
|
def _count_failed(sync_engine, source_id):
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
return s.execute(
|
|
select(func.count(PatreonFailedMedia.id)).where(
|
|
PatreonFailedMedia.source_id == source_id
|
|
)
|
|
).scalar_one()
|
|
|
|
|
|
def _seed_failed(sync_engine, source_id, key, attempts):
|
|
factory = sessionmaker(sync_engine, expire_on_commit=False)
|
|
with factory() as s:
|
|
s.add(PatreonFailedMedia(source_id=source_id, filehash=key, attempts=attempts))
|
|
s.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_repeated_failures_dead_letter_then_skip(source_id, sync_engine, tmp_path):
|
|
"""A media that errors DEAD_LETTER_THRESHOLD times is recorded; the next walk
|
|
skips it (no download attempt) and counts it dead-lettered."""
|
|
from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD
|
|
|
|
m1 = _media("p1", 1)
|
|
|
|
def _run():
|
|
ing = _ingester(
|
|
sync_engine, tmp_path,
|
|
_FakeClient([(None, [("p1", [m1])])]),
|
|
_FakeDownloader(tmp_path, error={_ledger_key(m1)}),
|
|
)
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
return result, ing.downloader
|
|
|
|
# Fail it up to the threshold — each run attempts the download and errors.
|
|
for _ in range(DEAD_LETTER_THRESHOLD):
|
|
result, dl = _run()
|
|
assert dl.download_calls == 1
|
|
assert _count_failed(sync_engine, source_id) == 1
|
|
|
|
# Now attempts == threshold → dead: the next walk skips it without trying.
|
|
result, dl = _run()
|
|
assert dl.download_calls == 0
|
|
assert result.run_stats["dead_lettered_count"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_reattempts_dead_lettered_and_clears_on_success(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Recovery bypasses the dead-letter ledger (re-attempts dead media); a clean
|
|
download clears the row (it recovered)."""
|
|
from backend.app.services.patreon_ingester import DEAD_LETTER_THRESHOLD
|
|
|
|
m1 = _media("p1", 1)
|
|
_seed_failed(sync_engine, source_id, _ledger_key(m1), DEAD_LETTER_THRESHOLD)
|
|
|
|
downloader = _FakeDownloader(tmp_path) # succeeds
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader)
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recovery",
|
|
)
|
|
assert downloader.download_calls == 1 # re-attempted despite being dead
|
|
assert result.files_downloaded == 1
|
|
assert _count_failed(sync_engine, source_id) == 0 # cleared on success
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_clean_download_clears_below_threshold_failure(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""A media with a sub-threshold failure that now downloads cleanly (tick) has
|
|
its dead-letter row cleared."""
|
|
m1 = _media("p1", 1)
|
|
_seed_failed(sync_engine, source_id, _ledger_key(m1), 1) # not yet dead
|
|
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, _FakeClient([(None, [("p1", [m1])])]), downloader)
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
assert result.files_downloaded == 1
|
|
assert _count_failed(sync_engine, source_id) == 0
|
|
|
|
|
|
# --- ledger + drift -------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_mark_seen_is_idempotent(source_id, sync_engine, tmp_path):
|
|
m1 = _media("p1", 1)
|
|
client = _FakeClient([(None, [("p1", [m1])])])
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
# Same media downloaded twice across two runs — the second marks seen again.
|
|
ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
|
|
ing._mark_seen(source_id, [(_ledger_key(m1), "p1")])
|
|
assert _count_ledger(sync_engine, source_id) == 1
|
|
|
|
|
|
def _run_with_client_error(source_id, sync_engine, tmp_path, exc):
|
|
client = _FakeClient([], raise_exc=exc)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
return ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_drift_maps_to_api_drift_and_fails_loud(source_id, sync_engine, tmp_path):
|
|
result = _run_with_client_error(
|
|
source_id, sync_engine, tmp_path,
|
|
PatreonDriftError("missing top-level 'data' key"),
|
|
)
|
|
assert result.success is False
|
|
assert result.error_type == ErrorType.API_DRIFT
|
|
assert "Patreon API changed" in (result.error_message or "")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_error_maps_to_auth_error(source_id, sync_engine, tmp_path):
|
|
result = _run_with_client_error(
|
|
source_id, sync_engine, tmp_path,
|
|
PatreonAuthError("HTTP 403 — auth rejected", status_code=403),
|
|
)
|
|
assert result.success is False
|
|
assert result.error_type == ErrorType.AUTH_ERROR
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_status_maps_to_rate_limited(source_id, sync_engine, tmp_path):
|
|
result = _run_with_client_error(
|
|
source_id, sync_engine, tmp_path,
|
|
PatreonAPIError("HTTP 429", status_code=429),
|
|
)
|
|
assert result.error_type == ErrorType.RATE_LIMITED
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_not_found_status_maps_to_not_found(source_id, sync_engine, tmp_path):
|
|
result = _run_with_client_error(
|
|
source_id, sync_engine, tmp_path,
|
|
PatreonAPIError("HTTP 404", status_code=404),
|
|
)
|
|
assert result.error_type == ErrorType.NOT_FOUND
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_transport_error_maps_to_network_error(source_id, sync_engine, tmp_path):
|
|
result = _run_with_client_error(
|
|
source_id, sync_engine, tmp_path,
|
|
PatreonAPIError("connection reset"), # no status_code → transport
|
|
)
|
|
assert result.error_type == ErrorType.NETWORK_ERROR
|
|
|
|
|
|
def test_ledger_key_video_has_no_filehash():
|
|
video = MediaItem(
|
|
url="https://stream.mux.com/abc.m3u8", filename="clip.mp4",
|
|
kind="postfile", filehash=None, post_id="p9",
|
|
)
|
|
assert _ledger_key(video) == "p9:clip.mp4"
|
|
|
|
|
|
# --- verify_patreon_credential (the verify counterpart, plan #697) ---------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_patreon_credential_unresolvable_is_inconclusive():
|
|
from backend.app.services.patreon_ingester import verify_patreon_credential
|
|
|
|
# No campaign id resolvable from a non-Patreon URL → can't test → None.
|
|
ok, msg = await verify_patreon_credential("not-a-patreon-url", None, {})
|
|
assert ok is None
|
|
assert "campaign id" in msg.lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_verify_patreon_credential_delegates_to_client(monkeypatch):
|
|
import backend.app.services.patreon_ingester as mod
|
|
|
|
async def _fake_resolve(url, cookies_path, overrides):
|
|
return "123", None
|
|
monkeypatch.setattr(mod, "resolve_campaign_id_for_source", _fake_resolve)
|
|
monkeypatch.setattr(
|
|
mod.PatreonClient, "verify_auth",
|
|
lambda self, campaign_id: (True, f"ok:{campaign_id}"),
|
|
)
|
|
ok, msg = await mod.verify_patreon_credential("https://patreon.com/x", None, {})
|
|
assert ok is True
|
|
assert msg == "ok:123"
|
|
|
|
|
|
# --- media-less (text-only) post capture ----------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_tick_captures_media_less_post_once(source_id, sync_engine, tmp_path):
|
|
"""A post with NO downloadable media still gets a post-only record, gated by
|
|
the seen-ledger (synthetic post key) so a second walk doesn't re-record it."""
|
|
client = _FakeClient([(None, [("ptext", [])])])
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
assert result.success is True
|
|
assert len(result.post_record_paths) == 1
|
|
assert downloader.post_records == 1
|
|
# The synthetic `post:ptext` key was marked seen (gates re-capture).
|
|
assert _count_ledger(sync_engine, source_id) == 1
|
|
|
|
# Second walk: already recorded → gated, no re-write, no new ledger row.
|
|
client2 = _FakeClient([(None, [("ptext", [])])])
|
|
downloader2 = _FakeDownloader(tmp_path)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
|
|
result2 = ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick",
|
|
)
|
|
assert result2.post_record_paths == []
|
|
assert downloader2.post_records == 0
|
|
assert _count_ledger(sync_engine, source_id) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_recovery_bypasses_gate_and_re_records(source_id, sync_engine, tmp_path):
|
|
"""Recovery bypasses the seen-ledger, so a media-less post is re-recorded
|
|
(matches recovery semantics for media)."""
|
|
client = _FakeClient([(None, [("ptext", [])])])
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
ing.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick")
|
|
|
|
client2 = _FakeClient([(None, [("ptext", [])])])
|
|
dl2 = _FakeDownloader(tmp_path)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, dl2)
|
|
result = ing2.run(source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="recovery")
|
|
assert len(result.post_record_paths) == 1
|
|
assert dl2.post_records == 1
|
|
|
|
|
|
# --- post-body schema-drift canary (#862) --------------------------------
|
|
|
|
|
|
def _media_less_posts(n):
|
|
# A single page of n media-less posts; each still gets a post-record, so the
|
|
# canary counts all n. Unique ids keep the seen-ledger gate from collapsing them.
|
|
return [(None, [(f"c{i}", []) for i in range(n)])]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_body_canary_fails_run_when_all_bodies_empty(source_id, sync_engine, tmp_path):
|
|
"""#862: a native walk that records >= _CANARY_MIN_SAMPLE posts but extracts a
|
|
body from NONE of them is the signature of a Patreon body-field rename (as
|
|
content→content_json_string was) — fail the run RED (API_DRIFT) instead of
|
|
silently archiving empties."""
|
|
client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE), empty_body=True)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert result.success is False
|
|
assert result.error_type == ErrorType.API_DRIFT
|
|
assert "canary" in (result.error_message or "").lower()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_body_canary_silent_when_some_bodies_present(source_id, sync_engine, tmp_path):
|
|
"""No false alarm: a large walk that DOES capture bodies completes normally
|
|
(normal operation — e.g. a gallery creator who writes captions)."""
|
|
client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE)) # bodies present
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert result.success is True
|
|
assert result.error_type is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_body_canary_silent_below_min_sample(source_id, sync_engine, tmp_path):
|
|
"""Tick safety: a small walk (a few new captionless posts) is below the sample
|
|
floor and must NOT trip the canary, even with every body empty."""
|
|
client = _FakeClient(_media_less_posts(_CANARY_MIN_SAMPLE - 1), empty_body=True)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill",
|
|
)
|
|
assert result.success is True
|
|
assert result.error_type is None
|
|
|
|
|
|
# --- the revisit window: reaching posts that were EDITED after capture -------
|
|
#
|
|
# Operator, 2026-09-23, on a Floppystack post: *"this post has been updated as
|
|
# he implements hot fixes — any chance we have a way to scan for or see updated
|
|
# posts so we can update ours to match and pull the new attachments and
|
|
# pictures etc."*
|
|
#
|
|
# A tick stopped after N contiguous already-have-it items, so an edit to a
|
|
# three-day-old post was structurally unreachable: it sits well below twenty
|
|
# seen items. The early-out now needs BOTH that run AND a post published before
|
|
# the horizon.
|
|
|
|
|
|
def _iso(days_ago):
|
|
return (datetime.now(UTC) - timedelta(days=days_ago)).isoformat()
|
|
|
|
|
|
def _seed_all_seen(sync_engine, source_id, media):
|
|
"""Several media through the module's existing one-key `_seed_seen`.
|
|
|
|
A distinct NAME, not a second definition: the first version of this shadowed
|
|
`_seed_seen` from the bottom of the file, and the two #830 recapture tests
|
|
that pass it a `post_id=` kwarg broke on a TypeError. A helper defined below
|
|
the tests that use it is invisible at the point of use.
|
|
"""
|
|
for m in media:
|
|
_seed_seen(sync_engine, source_id, _ledger_key(m), post_id=m.post_id)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_run_of_seen_items_does_not_stop_a_tick_inside_the_window(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The bug itself. Three all-seen posts, threshold 2 — the old walk turned
|
|
around at the second and never looked at the third, which is exactly where
|
|
an edited post lives."""
|
|
seen = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
_seed_all_seen(sync_engine, source_id, seen)
|
|
|
|
client = _FakeClient(
|
|
[(None, [(m.post_id, [m]) for m in seen])],
|
|
published={m.post_id: _iso(3) for m in seen},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
|
|
revisit_days=30,
|
|
)
|
|
|
|
assert result.success is True
|
|
assert client.consumed_posts == 3
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_the_early_out_still_fires_once_the_walk_is_below_the_horizon(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The other half, and the one that keeps a tick cheap. Same three posts,
|
|
published outside the window → the count early-out stands exactly as it
|
|
did. Asserted beside the test above because the window is only correct if
|
|
BOTH conditions are required; either one alone is a different feature."""
|
|
seen = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
_seed_all_seen(sync_engine, source_id, seen)
|
|
|
|
client = _FakeClient(
|
|
[(None, [(m.post_id, [m]) for m in seen])],
|
|
published={m.post_id: _iso(90) for m in seen},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
|
|
revisit_days=30,
|
|
)
|
|
|
|
assert result.success is True
|
|
assert client.consumed_posts == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_window_of_zero_is_the_behaviour_that_shipped_before_it(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The off switch. Recent posts, a window of 0 → the walk stops on the
|
|
count alone, so an operator who wants the old cheap tick has one."""
|
|
seen = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
_seed_all_seen(sync_engine, source_id, seen)
|
|
|
|
client = _FakeClient(
|
|
[(None, [(m.post_id, [m]) for m in seen])],
|
|
published={m.post_id: _iso(1) for m in seen},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
|
|
revisit_days=0,
|
|
)
|
|
|
|
assert client.consumed_posts == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_backfill_ignores_the_window_because_it_never_early_outs(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Stated so the window cannot quietly acquire a second job. A backfill
|
|
walks to the bottom regardless, and `recapture` is the mode that re-reads
|
|
every body — a horizon there would be a third answer to a two-answer
|
|
question."""
|
|
seen = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
_seed_all_seen(sync_engine, source_id, seen)
|
|
|
|
client = _FakeClient(
|
|
[(None, [(m.post_id, [m]) for m in seen])],
|
|
published={m.post_id: _iso(90) for m in seen},
|
|
)
|
|
downloader = _FakeDownloader(tmp_path)
|
|
ing = _ingester(sync_engine, tmp_path, client, downloader)
|
|
|
|
ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="backfill", seen_threshold=2,
|
|
revisit_days=30,
|
|
)
|
|
|
|
assert client.consumed_posts == 3
|
|
assert downloader.post_revisits == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_an_edited_post_inside_the_window_downloads_its_new_attachment(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The operator's case, end to end.
|
|
|
|
Walk one captures a post with one file. The creator then edits it to attach
|
|
a hotfix build. Walk two must download that file AND name the post as
|
|
updated — the ask was to SEE which posts changed, not only to end up with
|
|
their bytes.
|
|
|
|
The detection half needed nothing new: `extract_media` reads the media list
|
|
off the LIVE feed response every walk, so a newly attached file is simply a
|
|
ledger key we have never seen. Only reaching the post was missing.
|
|
"""
|
|
original = _media("p1", 1)
|
|
client1 = _FakeClient(
|
|
[(None, [("p1", [original])])], published={"p1": _iso(3)},
|
|
)
|
|
ing1 = _ingester(sync_engine, tmp_path, client1, _FakeDownloader(tmp_path))
|
|
ing1.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
hotfix = _media("p1", 2)
|
|
client2 = _FakeClient(
|
|
[(None, [("p1", [original, hotfix])])], published={"p1": _iso(3)},
|
|
)
|
|
downloader2 = _FakeDownloader(tmp_path)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
|
|
result = ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
assert result.files_downloaded == 1
|
|
assert "1 post(s) updated (1 new file(s))" in result.stdout
|
|
assert "post p1 — updated: 1 new file(s)" in result.stdout
|
|
# The post record was re-read, not skipped by the capture gate.
|
|
assert downloader2.post_revisits == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_post_below_the_horizon_keeps_its_capture_gate(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""The gate still does its job everywhere the window does not reach — else
|
|
the window would have quietly become "re-read every post forever"."""
|
|
old = _media("pold", 1)
|
|
client1 = _FakeClient([(None, [("pold", [old])])], published={"pold": _iso(90)})
|
|
ing1 = _ingester(sync_engine, tmp_path, client1, _FakeDownloader(tmp_path))
|
|
ing1.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
client2 = _FakeClient([(None, [("pold", [old])])], published={"pold": _iso(90)})
|
|
downloader2 = _FakeDownloader(tmp_path)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
|
|
ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
assert downloader2.post_records == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_revisit_that_finds_nothing_new_is_not_reported_as_an_update(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""Most revisits find nothing — that is the normal case, and a run summary
|
|
claiming "1 post(s) updated" on every tick would train the operator to stop
|
|
reading it (lesson: a signal that is always on is not a signal)."""
|
|
m = _media("p1", 1)
|
|
client1 = _FakeClient([(None, [("p1", [m])])], published={"p1": _iso(3)})
|
|
ing1 = _ingester(sync_engine, tmp_path, client1, _FakeDownloader(tmp_path))
|
|
ing1.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
client2 = _FakeClient([(None, [("p1", [m])])], published={"p1": _iso(3)})
|
|
downloader2 = _FakeDownloader(tmp_path)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, downloader2)
|
|
result = ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
assert result.files_downloaded == 0
|
|
assert "updated" not in result.stdout
|
|
assert downloader2.post_revisits == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_revisits_do_not_feed_the_body_drift_canary(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""#862's canary fails a run that recorded a meaningful sample of posts and
|
|
got a body from NONE of them. A revisit legitimately comes back empty — the
|
|
post's body only ever arrived from the detail endpoint, which a revisit
|
|
declines to call — so counting revisits into that sample would walk the
|
|
alarm toward firing on healthy ticks. First captures only."""
|
|
posts = [(f"c{i}", []) for i in range(_CANARY_MIN_SAMPLE)]
|
|
published = {pid: _iso(3) for pid, _ in posts}
|
|
|
|
client1 = _FakeClient([(None, posts)], published=published)
|
|
ing1 = _ingester(sync_engine, tmp_path, client1, _FakeDownloader(tmp_path))
|
|
first = ing1.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
assert first.success is True
|
|
|
|
# Second walk: every post is a revisit, and every body comes back empty.
|
|
client2 = _FakeClient([(None, posts)], published=published, empty_body=True)
|
|
ing2 = _ingester(sync_engine, tmp_path, client2, _FakeDownloader(tmp_path))
|
|
second = ing2.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", revisit_days=30,
|
|
)
|
|
|
|
assert second.success is True
|
|
assert second.error_type is not ErrorType.API_DRIFT
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_a_post_whose_date_will_not_parse_falls_back_to_the_count(
|
|
source_id, sync_engine, tmp_path,
|
|
):
|
|
"""A date we cannot read must not fail the walk, and must not be guessed
|
|
into the window. It reads as "not provably recent", which leaves that post
|
|
on the behaviour it had before the window existed."""
|
|
seen = [_media(f"p{i}", 1) for i in range(1, 4)]
|
|
_seed_all_seen(sync_engine, source_id, seen)
|
|
|
|
client = _FakeClient(
|
|
[(None, [(m.post_id, [m]) for m in seen])],
|
|
published={m.post_id: "last Tuesday" for m in seen},
|
|
)
|
|
ing = _ingester(sync_engine, tmp_path, client, _FakeDownloader(tmp_path))
|
|
|
|
result = ing.run(
|
|
source_id=source_id, campaign_id="c1", artist_slug="ingest",
|
|
url="https://patreon.com/ingest", mode="tick", seen_threshold=2,
|
|
revisit_days=30,
|
|
)
|
|
|
|
assert result.success is True
|
|
assert client.consumed_posts == 2
|