` 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 `
'
'
May 01, 2026 12:00 pm
'
'
'
)
page = (
'
'
f'{raw_post}
'
)
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: "nothing")
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 = (
'
'
)
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="
text only
")))
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="
real body
"), next_href="/p?page=2")
page = page.replace(
'
',
'
'
"View next posts (264 / 265)
",
)
[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 inner (gallery-dl parity).
client = SubscribeStarClient(None)
wrapped = "\n
hello world
\n\n"
[parsed] = client._parse_posts(_feed_page(_post_html("701", body=wrapped)))
assert parsed["attributes"]["content"] == "
hello world
"
# -- 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": "
body text
",
"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 → "
__" (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