bvandeusen
78a3977f8a
First live run (cheunart) tripped the drift guard: "no posts and no recognizable feed container". The browser-saved page was normal (6 posts + posts_container-list), so the parser was fine — our live HTTP fetch got a different response. Cause: the client set X-Requested-With: XMLHttpRequest (+ a JSON Accept) session-wide, so the initial creator-page GET was sent as an XHR. SubscribeStar (Rails) content- negotiates an XHR full-page request to a non-HTML body → no container → drift. Fix: the session now uses browser-like navigation headers (Accept: html, NO X-Requested-With); the XHR header + JSON Accept are applied PER-REQUEST only on the "load more" endpoint (which is a genuine XHR). Drift message now reports the response length + a JSON hint so a recurrence is self-explaining. Regression test pins the header split (navigation initial GET, XHR load-more). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""Unit tests for the native SubscribeStar client / downloader / ingester.
|
|
|
|
No network, no DB (PURE modules → fast lane, unmarked). The HTML feed is fed via
|
|
monkeypatched `_feed_html` / `_loadmore_html`; the media HTTP layer via a fake
|
|
`session` seam. Canned HTML mirrors the real markup characterized in the Step-0
|
|
spike (Scribe note "SubscribeStar HTML characterization").
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
|
|
from backend.app.services.gallery_dl import DownloadResult, ErrorType
|
|
from backend.app.services.native_ingest_common import post_dir_name
|
|
from backend.app.services.subscribestar_client import (
|
|
MediaItem,
|
|
SubscribeStarAPIError,
|
|
SubscribeStarAuthError,
|
|
SubscribeStarClient,
|
|
SubscribeStarDriftError,
|
|
_parse_ss_datetime,
|
|
)
|
|
from backend.app.services.subscribestar_downloader import SubscribeStarDownloader
|
|
from backend.app.services.subscribestar_ingester import (
|
|
SubscribeStarIngester,
|
|
_ledger_key,
|
|
)
|
|
from backend.app.utils.sidecar import find_sidecar
|
|
|
|
_PNG_HEAD = b"\x89PNG\r\n\x1a\n"
|
|
_PNG_BYTES = _PNG_HEAD + (b"\x00" * 16) + b"\x00\x00\x00\x00IEND\xaeB`\x82"
|
|
|
|
|
|
def _gallery_attr(items: list[dict]) -> str:
|
|
# data-gallery is HTML-entity-escaped JSON.
|
|
return json.dumps(items).replace("&", "&").replace('"', """)
|
|
|
|
|
|
def _post_html(post_id="111", date="May 01, 2026 12:00 pm", body="<p>hello</p>",
|
|
media: list[dict] | None = None, locked=False):
|
|
gallery = ""
|
|
if media is not None:
|
|
gallery = (
|
|
f'<div class="uploads-images" data-gallery="{_gallery_attr(media)}"></div>'
|
|
)
|
|
lock = ' for-locked_content' if locked else ""
|
|
return (
|
|
f'<div class="post is-shown false false{lock}" data-id="{post_id}" '
|
|
f'data-infinite-scroll-id="{post_id}">'
|
|
f'<div class="post-date">{date}</div>'
|
|
f'<div class="post-body" data-role="post-body">'
|
|
f'<div class="post-content" data-role="post_content-text">'
|
|
f'<div class="trix-content">{body}</div></div></div>'
|
|
f'{gallery}</div>'
|
|
)
|
|
|
|
|
|
def _feed_page(posts_html: str, next_href: str | None = None):
|
|
nxt = (
|
|
f'<div data-role="infinite_scroll-next_page" href="{next_href}"></div>'
|
|
if next_href else ""
|
|
)
|
|
return (
|
|
'<html><body><div data-role="posts_container-list">'
|
|
f'{posts_html}{nxt}</div></body></html>'
|
|
)
|
|
|
|
|
|
# -- client: dates -----------------------------------------------------------
|
|
|
|
def test_parse_ss_datetime_variants():
|
|
assert _parse_ss_datetime("Jun 17, 2026 03:19 am") == "2026-06-17T03:19:00"
|
|
assert _parse_ss_datetime("Updated on Jul 11, 2025 02:39 pm") == "2025-07-11T14:39:00"
|
|
assert _parse_ss_datetime("nonsense") is None
|
|
|
|
|
|
# -- client: request headers (regression: live-fetch drift) -----------------
|
|
|
|
def test_initial_fetch_is_navigation_loadmore_is_xhr(monkeypatch):
|
|
# The initial creator-page GET must NOT be flagged XHR (Rails content-
|
|
# negotiates an XHR full-page request to a non-HTML body → drift; first live
|
|
# run, cheunart 2026-06-17). Only the "load more" endpoint is a real XHR.
|
|
client = SubscribeStarClient(None)
|
|
seen = []
|
|
|
|
class _R:
|
|
status_code = 200
|
|
text = '<div data-role="posts_container-list"></div>'
|
|
|
|
def json(self):
|
|
return {"html": ""}
|
|
|
|
def fake_get(url, *, headers=None):
|
|
seen.append(headers)
|
|
return _R()
|
|
|
|
monkeypatch.setattr(client, "_get", fake_get)
|
|
client._feed_html("https://subscribestar.adult/x")
|
|
client._loadmore_html("https://subscribestar.adult/posts?page=2")
|
|
assert seen[0] is None # navigation — no per-request XHR header
|
|
assert seen[1]["X-Requested-With"] == "XMLHttpRequest"
|
|
|
|
|
|
# -- client: iteration + pagination -----------------------------------------
|
|
|
|
def test_iter_posts_parses_and_paginates(monkeypatch):
|
|
client = SubscribeStarClient(None)
|
|
page1 = _feed_page(
|
|
_post_html("111") + _post_html("112"),
|
|
next_href="/posts?page=2&slug=x&sort_by=newest",
|
|
)
|
|
page2 = _feed_page(_post_html("113")) # no next link → stop
|
|
monkeypatch.setattr(client, "_feed_html", lambda url: page1)
|
|
monkeypatch.setattr(client, "_loadmore_html", lambda url: page2)
|
|
|
|
out = list(client.iter_posts("https://subscribestar.adult/x"))
|
|
ids = [p["id"] for p, _inc, _cur in out]
|
|
assert ids == ["111", "112", "113"]
|
|
# page_cursor is the value that FETCHED each post's page: None for page 1, the
|
|
# next_page href for page 2.
|
|
assert out[0][2] is None
|
|
assert out[2][2] == "/posts?page=2&slug=x&sort_by=newest"
|
|
# base is stamped for media URL joining.
|
|
assert out[0][0]["_base"] == "https://subscribestar.adult"
|
|
|
|
|
|
def test_iter_posts_drift_when_no_container(monkeypatch):
|
|
client = SubscribeStarClient(None)
|
|
monkeypatch.setattr(client, "_feed_html", lambda url: "<html>nothing</html>")
|
|
try:
|
|
list(client.iter_posts("https://subscribestar.adult/x"))
|
|
except SubscribeStarDriftError:
|
|
return
|
|
raise AssertionError("expected SubscribeStarDriftError")
|
|
|
|
|
|
# -- client: media extraction ------------------------------------------------
|
|
|
|
def test_extract_media_from_gallery():
|
|
client = SubscribeStarClient(None)
|
|
media = [
|
|
{"id": 9001, "type": "image", "original_filename": "art.jpg",
|
|
"url": "/post_uploads?payload=ABC"},
|
|
{"id": 9002, "type": "image", "original_filename": "art2.jpg",
|
|
"url": "/post_uploads?payload=DEF"},
|
|
]
|
|
[post] = client._parse_posts(_feed_page(_post_html("111", media=media)))
|
|
post["_base"] = "https://subscribestar.adult"
|
|
items = client.extract_media(post, {})
|
|
assert [m.media_id for m in items] == ["9001", "9002"]
|
|
assert items[0].url == "https://subscribestar.adult/post_uploads?payload=ABC"
|
|
assert items[0].filename == "art.jpg"
|
|
assert items[0].post_id == "111"
|
|
|
|
|
|
def test_text_post_has_no_media_but_keeps_body():
|
|
client = SubscribeStarClient(None)
|
|
[post] = client._parse_posts(_feed_page(_post_html("111", body="<p>text only</p>")))
|
|
post["_base"] = "https://subscribestar.adult"
|
|
assert client.extract_media(post, {}) == []
|
|
assert "text only" in post["attributes"]["content"]
|
|
|
|
|
|
# -- client: gating + record key --------------------------------------------
|
|
|
|
def test_post_is_gated():
|
|
client = SubscribeStarClient(None)
|
|
[open_post] = client._parse_posts(_feed_page(_post_html("1", media=[])))
|
|
[locked] = client._parse_posts(_feed_page(_post_html("2", locked=True)))
|
|
assert client.post_is_gated(open_post) is False
|
|
assert client.post_is_gated(locked) is True
|
|
|
|
|
|
def test_post_record_key():
|
|
assert SubscribeStarClient.post_record_key({"id": "55"}) == ("post:55", "55")
|
|
assert SubscribeStarClient.post_record_key({}) is None
|
|
|
|
|
|
# -- downloader --------------------------------------------------------------
|
|
|
|
class _FakeResponse:
|
|
def __init__(self, payload=_PNG_BYTES, status_code=200):
|
|
self._payload = payload
|
|
self.status_code = status_code
|
|
self.headers: dict = {}
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise requests.HTTPError(f"HTTP {self.status_code}")
|
|
|
|
def iter_content(self, chunk_size=65536):
|
|
for i in range(0, len(self._payload), chunk_size):
|
|
yield self._payload[i:i + chunk_size]
|
|
|
|
|
|
class _FakeSession:
|
|
def __init__(self):
|
|
self.calls: list[str] = []
|
|
|
|
def get(self, url, stream=False, timeout=None, headers=None):
|
|
self.calls.append(url)
|
|
return _FakeResponse()
|
|
|
|
|
|
def _post(post_id="111", date="May 01, 2026 12:00 pm"):
|
|
return {
|
|
"id": post_id,
|
|
"attributes": {
|
|
"title": "",
|
|
"content": "<div>body text</div>",
|
|
"published_at": _parse_ss_datetime(date),
|
|
"post_type": "subscribestar",
|
|
},
|
|
}
|
|
|
|
|
|
def _img(name, post_id="111", media_id="9001"):
|
|
return MediaItem(
|
|
url=f"https://subscribestar.adult/post_uploads?payload={name}",
|
|
filename=name, kind="image", filehash=None,
|
|
post_id=post_id, media_id=media_id,
|
|
)
|
|
|
|
|
|
def _downloader(tmp_path: Path, session=None, validate=True):
|
|
return SubscribeStarDownloader(
|
|
images_root=tmp_path, cookies_path=None, validate=validate,
|
|
session=session or _FakeSession(),
|
|
)
|
|
|
|
|
|
def test_post_dir_name_empty_title_matches_gallery_dl():
|
|
# SubscribeStar has no title → "<date>_<id>_" (matches gallery-dl's layout so
|
|
# existing downloads dedup on disk at cutover).
|
|
assert post_dir_name(_post("111")) == "2026-05-01_111_"
|
|
|
|
|
|
def test_download_post_layout_and_outcomes(tmp_path):
|
|
# .png names so the PNG bytes pass the content/extension validator.
|
|
dl = _downloader(tmp_path)
|
|
outcomes = dl.download_post(_post(), [_img("a.png"), _img("b.png", media_id="9002")], "artist-x")
|
|
post_dir = tmp_path / "artist-x" / "subscribestar" / "2026-05-01_111_"
|
|
assert (post_dir / "01_a.png").is_file()
|
|
assert (post_dir / "02_b.png").is_file()
|
|
assert [o.status for o in outcomes] == ["downloaded", "downloaded"]
|
|
|
|
|
|
def test_per_media_sidecar_is_minimal(tmp_path):
|
|
dl = _downloader(tmp_path)
|
|
outcomes = dl.download_post(_post(), [_img("a.png")], "artist-x")
|
|
sidecar = find_sidecar(outcomes[0].path)
|
|
data = json.loads(sidecar.read_text())
|
|
assert data["category"] == "subscribestar"
|
|
assert data["id"] == "111"
|
|
assert data["source_url"].endswith("payload=a.png")
|
|
# post-first: no body/title in the per-media sidecar.
|
|
assert "content" not in data and "title" not in data
|
|
|
|
|
|
def test_write_post_record(tmp_path):
|
|
dl = _downloader(tmp_path)
|
|
rec = dl.write_post_record(_post("111"), "artist-x")
|
|
assert rec.path.name == "_post.json"
|
|
data = json.loads(rec.path.read_text())
|
|
assert data["category"] == "subscribestar"
|
|
assert data["id"] == "111" and data["post_id"] == "111"
|
|
assert "body text" in data["content"]
|
|
assert rec.body_chars > 0
|
|
|
|
|
|
def test_skip_seen_does_not_download(tmp_path):
|
|
session = _FakeSession()
|
|
dl = _downloader(tmp_path, session=session)
|
|
outcomes = dl.download_post(
|
|
_post(), [_img("a.jpg")], "artist-x", is_seen=lambda m: True,
|
|
)
|
|
assert [o.status for o in outcomes] == ["skipped_seen"]
|
|
assert session.calls == []
|
|
|
|
|
|
# -- ingester ----------------------------------------------------------------
|
|
|
|
def test_ledger_key_prefers_filehash_then_post_media():
|
|
assert _ledger_key(_img("a.jpg")) == "111:9001" # no filehash → post:media
|
|
hashed = MediaItem(
|
|
url="u", filename="a.jpg", kind="image", filehash="deadbeef",
|
|
post_id="111", media_id="9001",
|
|
)
|
|
assert _ledger_key(hashed) == "deadbeef"
|
|
|
|
|
|
def _ingester(tmp_path):
|
|
return SubscribeStarIngester(
|
|
images_root=tmp_path, cookies_path=None, session_factory=lambda: None,
|
|
)
|
|
|
|
|
|
def _mk_result(**kw) -> DownloadResult:
|
|
# _failure_result supplies success/return_code/error_type/error_message in kw;
|
|
# the factory only injects the always-required identity fields.
|
|
return DownloadResult(url="u", artist_slug="a", platform="subscribestar", **kw)
|
|
|
|
|
|
def test_failure_result_maps_exceptions(tmp_path):
|
|
ing = _ingester(tmp_path)
|
|
assert ing._failure_result(SubscribeStarAuthError("x"), _mk_result).error_type == ErrorType.AUTH_ERROR
|
|
assert ing._failure_result(SubscribeStarDriftError("x"), _mk_result).error_type == ErrorType.API_DRIFT
|
|
assert ing._failure_result(
|
|
SubscribeStarAPIError("x", status_code=429), _mk_result
|
|
).error_type == ErrorType.RATE_LIMITED
|
|
assert ing._failure_result(
|
|
SubscribeStarAPIError("x", status_code=404), _mk_result
|
|
).error_type == ErrorType.NOT_FOUND
|
|
assert ing._failure_result(
|
|
SubscribeStarAPIError("x"), _mk_result
|
|
).error_type == ErrorType.NETWORK_ERROR
|