bvandeusen/FabledCurator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 33 Wiki Activity
Files
8f6547f8d7d8ecfefeee9c6e873310106a97c2b2
FabledCurator/tests/test_subscribestar_native.py
T
bvandeusen 0963bf0db3
CI / lint (push) Successful in 2s
Details
CI / frontend-build (push) Successful in 20s
Details
CI / backend-lint-and-test (push) Successful in 32s
Details
CI / integration (push) Successful in 3m41s
Details
feat(artist): resolve patreon + subscribestar display names at add-time (#130 step 5) 
Parity with pixiv (operator ask): the extension add now resolves the real
display name for our other native platforms too, not just the URL handle.
patreon_resolver.resolve_display_name reads the campaigns API's
attributes.name; SubscribeStarClient.resolve_display_name pulls the creator
name off the profile page (og:title, else the <title> stripped of the
SubscribeStar suffix). extension_service._resolve_artist_name dispatches per
platform (pixiv=token, patreon/subscribestar=cookies via get_cookies_path),
best-effort in an executor, falling back to the readable URL handle on any
failure. Still all curator core — the extension is unchanged (sends only the
URL). gallery-dl platforms keep the handle (readable, no native client).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CDgx8bQS5YrGRK76v8HUnM
2026-07-04 22:37:50 -04:00

502 lines
20 KiB
Python
Raw Blame History

"""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,
_extract_creator_name,
_parse_ss_datetime,
_split_creator_url,
)
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("&", "&amp;").replace('"', "&quot;")
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>'
# The youtube-uploads div always follows post-content in real markup; it
# is the close marker our content extraction (and gallery-dl's) keys on.
f'<div class="post-uploads for-youtube" '
f'data-role="post_content-youtube_uploads"></div>'
f'</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
def test_split_creator_url_rewrites_art_to_adult():
# The .art age wall doesn't clear with the 18+ cookie; the same creator is
# reachable on .adult (Elasid, event #54116). Requests must route to .adult.
base, slug = _split_creator_url("https://subscribestar.art/elasid")
assert base == "https://subscribestar.adult"
assert slug == "elasid"
# www. prefix + trailing path still normalizes the host only.
base, slug = _split_creator_url("https://www.subscribestar.art/elasid/posts")
assert base == "https://www.subscribestar.adult"
assert slug == "elasid"
def test_split_creator_url_leaves_com_and_adult_untouched():
assert _split_creator_url("https://subscribestar.adult/sabu")[0] == \
"https://subscribestar.adult"
assert _split_creator_url("https://www.subscribestar.com/sabu")[0] == \
"https://www.subscribestar.com"
def test_extract_creator_name_prefers_og_title():
html = (
'<head><meta property="og:title" content="Sabu &amp; Friends">'
'<title>Sabu | SubscribeStar</title></head>'
)
assert _extract_creator_name(html) == "Sabu & Friends"
def test_extract_creator_name_title_fallback_strips_suffix():
assert _extract_creator_name(
"<head><title>Elasid on SubscribeStar</title></head>"
) == "Elasid"
assert _extract_creator_name(
"<head><title>Cheun | SubscribeStar — adult</title></head>"
) == "Cheun"
def test_extract_creator_name_none_when_absent():
assert _extract_creator_name("<html><body>no title</body></html>") is None
def test_date_parses_when_wrapped_in_permalink_anchor():
# Image posts wrap the date in an <a> permalink; a plain regex missed them
# → null dates (cheunart 2026-06-17). gallery-dl's method handles both.
client = SubscribeStarClient(None)
wrapped = (
'<div class="post is-shown" data-id="900">'
'<div class="post-date"><a href="/posts/900">May 01, 2026 12:00 pm</a></div>'
'<div class="post-content" data-role="post_content-text">'
'<div class="trix-content"><p>x</p></div></div>'
'<div class="post-uploads for-youtube"></div></div>'
)
[p] = client._parse_posts(wrapped)
assert p["attributes"]["published_at"] == "2026-05-01T12:00:00"
# -- client: request headers (regression: live-fetch drift) -----------------
def test_request_profile_matches_gallery_dl(monkeypatch):
# gallery-dl's default (cookies-only) mode: Firefox UA, Accept: */*, a same-
# site Referer (root/), and NO X-Requested-With on EITHER the creator page or
# the "load more" JSON endpoint. That profile is what avoids SubscribeStar's
# /verify_subscriber 302 with valid cookies (cheunart, 2026-06-17).
client = SubscribeStarClient(None)
assert "Firefox" in client._session.headers["User-Agent"]
assert client._session.headers["Accept"] == "*/*"
assert "X-Requested-With" not in client._session.headers
class _R:
status_code = 200
history: list = []
url = "https://subscribestar.adult/x"
text = '<div class="post " data-id="1"></div>'
def json(self):
return {"html": ""}
monkeypatch.setattr(client, "_get", lambda url, *, headers=None: _R())
list(client.iter_posts("https://subscribestar.adult/x"))
# Referer is stamped to the creator base for the walk.
assert client._session.headers["Referer"] == "https://subscribestar.adult/"
# -- 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_parses_raw_server_html_without_is_shown(monkeypatch):
# Regression (cheunart, 2026-06-17): the RAW server HTML opens posts as
# `<div class="post ...">` WITHOUT the `is-shown` class — that class is added
# by SubscribeStar's infinite-scroll JS when a post scrolls into view, so it's
# present in a browser-SAVED page but absent from the feed we actually fetch.
# The delimiter must match the generic `<div class="post ` (gallery-dl parity),
# else the live feed parses to zero posts → false drift.
client = SubscribeStarClient(None)
raw_post = (
'<div class="post for-subscribers" data-id="555" '
'data-infinite-scroll-id="555">'
'<div class="post-date">May 01, 2026 12:00 pm</div>'
'<div class="post-body" data-role="post-body">'
'<div class="post-content" data-role="post_content-text">'
'<div class="trix-content"><p>raw</p></div></div></div></div>'
)
page = (
'<html><body><div data-role="posts_container-list">'
f'{raw_post}</div></body></html>'
)
monkeypatch.setattr(client, "_feed_html", lambda url: page)
out = list(client.iter_posts("https://subscribestar.adult/x"))
assert [p["id"] for p, _inc, _cur in out] == ["555"]
def test_redirect_to_age_wall_is_auth_error():
# gallery-dl parity: SubscribeStar redirects an unauthenticated / age-
# unconfirmed request to /age_confirmation_warning (or /verify_subscriber).
# That's auth, not drift — rotate cookies / set the age cookie.
client = SubscribeStarClient(None)
class _Resp:
status_code = 200
url = "https://subscribestar.adult/age_confirmation_warning"
history = [object()]
text = ""
headers: dict = {}
class _Session:
def get(self, url, timeout=None, headers=None):
return _Resp()
client._session = _Session()
try:
client._get("https://subscribestar.adult/x")
except SubscribeStarAuthError:
return
raise AssertionError("expected SubscribeStarAuthError")
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_extract_media_skips_locked_previews():
# gallery-dl's _media_from_post skips gallery items under /previews (locked/
# blurred teasers) — the SubscribeStar gated-preview safeguard (#874).
client = SubscribeStarClient(None)
media = [
{"id": 1, "type": "image", "original_filename": "real.jpg",
"url": "/post_uploads?payload=ABC"},
{"id": 2, "type": "image", "original_filename": "locked.jpg",
"url": "/previews/abc/locked.jpg"},
]
[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] == ["1"]
def test_extract_media_includes_doc_and_audio_attachments():
# gallery-dl _media_from_post: document (uploads-docs/doc_preview, href) and
# audio (uploads-audios/audio_preview-data, src) attachments are separate from
# data-gallery — some posts deliver content ONLY through these.
client = SubscribeStarClient(None)
chunk = (
'<div class="post " data-id="888">'
'<div class="uploads-docs">'
'<div class="doc_preview" data-upload-id="501">'
'<a href="/post_uploads/d/chapter.pdf">'
'<div class="doc_preview-title">chapter.pdf</div></a></div></div>'
'<div class="uploads-audios">'
'<div class="audio_preview-data" data-upload-id="502">'
'<audio src="/post_uploads/a/track.mp3"></audio>'
'<div class="audio_preview-title">track.mp3</div></div></div>'
'<div class="post-edit_form"></div></div>'
)
post = {"id": "888", "_html": chunk, "_base": "https://subscribestar.adult"}
by_kind = {m.kind: m for m in client.extract_media(post, {})}
assert by_kind["attachment"].url == "https://subscribestar.adult/post_uploads/d/chapter.pdf"
assert by_kind["attachment"].filename == "chapter.pdf"
assert by_kind["attachment"].media_id == "501"
assert by_kind["audio"].url == "https://subscribestar.adult/post_uploads/a/track.mp3"
assert by_kind["audio"].filename == "track.mp3"
assert by_kind["audio"].media_id == "502"
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"]
def test_body_does_not_bleed_into_pagination_counter():
# Regression (cheunart 2026-06-17): the body must stop at the post's own
# content, not run into sibling upload divs or the "View next posts (N / M)"
# pagination counter (which rendered as a bogus "264 / 265" body/title).
client = SubscribeStarClient(None)
page = _feed_page(_post_html("700", body="<p>real body</p>"), next_href="/p?page=2")
page = page.replace(
'<div data-role="infinite_scroll-next_page" href="/p?page=2"></div>',
'<div data-role="infinite_scroll-next_page" href="/p?page=2">'
"View next posts (264 / 265)</div>",
)
[parsed] = client._parse_posts(page)
content = parsed["attributes"]["content"]
assert "real body" in content
assert "264 / 265" not in content
assert "View next posts" not in content
def test_body_strips_trix_html_document_wrapper():
# The trix editor serializes rich bodies as a full HTML document; keep only
# the <body> inner (gallery-dl parity).
client = SubscribeStarClient(None)
wrapped = "<!DOCTYPE html><html><body>\n<div>hello world</div>\n</body></html>\n"
[parsed] = client._parse_posts(_feed_page(_post_html("701", body=wrapped)))
assert parsed["attributes"]["content"] == "<div>hello world</div>"
# -- 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
Reference in New Issue View Git Blame Copy Permalink