"""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="

hello

", media: list[dict] | None = None, locked=False): gallery = "" if media is not None: gallery = ( f'

' ) lock = ' for-locked_content' if locked else "" return ( f'

' f'

{date}

' f'

{body}

' f'{gallery}

' ) def _feed_page(posts_html: str, next_href: str | None = None): nxt = ( f'

' if next_href else "" ) return ( '

' f'{posts_html}{nxt}

' ) # -- 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 = '

' 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: "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_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"] # -- 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