diff --git a/scripts/twitter/README.md b/scripts/twitter/README.md index 9d1d7a8..385f8b3 100644 --- a/scripts/twitter/README.md +++ b/scripts/twitter/README.md @@ -90,11 +90,8 @@ Later emits use that same title for the folder and the frontmatter. Move both si | `catbox_client.py` | one Catbox upload | | `fallback_media.py` | fallback activation and restore-origin | | `frozen.py` | resolve frozen ids against every captured post id | -| `lift_catbox.py` | retired. Use `tw.py fallback` | | `ocr_pass.py` | Umi HTTP `:1224` / tesseract / Windows OCR | | `media_embed.py` | `--attach-ocr` only | | `media_merge.py` | restore derived rows | -| `crt_pass.py` | stock slang only. ShaderGlass is the look reference | | `do_not_refetch.txt` | frozen id list | | `tests/` | isolated unit tests | -| `unfinished_bloat/` | leftover probe code. Not the front door | diff --git a/scripts/twitter/audit_handle_index.py b/scripts/twitter/audit_handle_index.py index b6ce3b4..fc84c2b 100644 --- a/scripts/twitter/audit_handle_index.py +++ b/scripts/twitter/audit_handle_index.py @@ -79,8 +79,7 @@ def audit_one(handle_dir: Path) -> dict: def audit_threads_index(threads_root: Path) -> dict: - """Audit the top-level archive/threads/index.md against real - handle dirs.""" + """Audit the top-level archive/threads/index.md against real handle dirs.""" idx_path = threads_root / "index.md" real = { d.name for d in threads_root.iterdir() if d.is_dir() and not d.name.startswith(".") diff --git a/scripts/twitter/crt_pass.py b/scripts/twitter/crt_pass.py deleted file mode 100644 index 2ca9f31..0000000 --- a/scripts/twitter/crt_pass.py +++ /dev/null @@ -1,322 +0,0 @@ -"""Apply a stock RetroArch slang preset to a still via librashader-cli. - -Does not patch shaders. Does not invent PixelSize. Your ShaderGlass .sgp files -are the look reference; this only runs the same .slangp with the listed params. - -Manual look you are matching: - 1. Nearest-upscale the orig (Affinity zoom). 300% => 3x. - 2. Overlay ShaderGlass with the .sgp (Cyclon PixelSize 6, Lottes PixelSize 4). - 3. Snapshot, then crop. - -This script can do step 1 (ffmpeg neighbor) and a stock slang render of the -result. It cannot reproduce ShaderGlass window capture. If the plate is wrong, -do it in ShaderGlass and drop the PNG next to the orig as *_crt.png. - -Presets: - lottes crt-lottes-multipass.slangp - LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp - cyclon crt-Cyclon.slangp - LOTTES_CRT_CODE_(CYCLON 6X).sgp - -auto: cyclon if max(w,h) <= --auto-max (default 720), else lottes. -""" -from __future__ import annotations - -import argparse -import json -import shutil -import subprocess -import sys -from pathlib import Path - -LOTTES_SGP: Path = Path( - r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" - r"\LOTTES_CODE_(LOTTES_MULTIPASS 2X) V2.sgp" -) -CYCLON_SGP: Path = Path( - r"C:\Users\Ed\Downloads\ShaderGlass-1.2.2-win-x64" - r"\LOTTES_CRT_CODE_(CYCLON 6X).sgp" -) -SLANG_LOTTES: Path = Path( - r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" - r"\shaders_slang\crt\crt-lottes-multipass.slangp" -) -SLANG_CYCLON: Path = Path( - r"C:\Users\Ed\scoop\apps\retroarch\1.22.2\shaders" - r"\shaders_slang\crt\crt-Cyclon.slangp" -) -CLI_DEFAULT: Path = Path(r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe") - -# Knobs copied from the .sgp files. Not a substitute for PixelSize. -LOTTES_PARAMS: str = ( - "hardScan=-19,hardPix=-7,warpX=0,warpY=0,maskDark=1.3,maskLight=1.3," - "shadowMask=4,brightBoost=1.15,bloomAmount=0,shape=1.9" -) -CYCLON_PARAMS: str = ( - "SCANLINE=0.2,INTERLACE=0,M_TYPE=0,SLOT=0,SLOTW=2,Maskl=0.55,Maskh=1," - "bzl=0,ambient=0,REFLECT=0,WARPX=0,WARPY=0,BR_DEP=0,BRIGHTNESS_=1.05," - "BLACK=0.01,vig=0" -) - - -def probe_size(path: Path) -> tuple[int, int]: - """Return ``(width, height)`` of an image or video via ``ffprobe``.""" - raw = subprocess.check_output( - [ - "ffprobe", - "-v", - "error", - "-select_streams", - "v:0", - "-show_entries", - "stream=width,height", - "-of", - "json", - str(path), - ], - text=True, - ) - stream = json.loads(raw)["streams"][0] - return int(stream["width"]), int(stream["height"]) - - -def pick_preset(width: int, height: int, name: str, auto_max: int) -> tuple[str, Path, Path, str]: - """Resolve the preset name to ``(name, .sgp_path, .slangp_path, params)``; ``auto`` picks cyclon under ``auto_max``.""" - if name == "auto": - name = "cyclon" if max(width, height) <= auto_max else "lottes" - if name == "lottes": - return name, LOTTES_SGP, SLANG_LOTTES, LOTTES_PARAMS - if name == "cyclon": - return name, CYCLON_SGP, SLANG_CYCLON, CYCLON_PARAMS - raise SystemExit(f"unknown preset {name}") - - -def find_cli(explicit: Path | None) -> Path: - """Locate ``librashader-cli``: explicit path, then PATH, then the scoop-installed default.""" - if explicit is not None: - if not explicit.is_file(): - raise SystemExit(f"missing --cli {explicit}") - return explicit - which = shutil.which("librashader-cli") - if which: - return Path(which) - if CLI_DEFAULT.is_file(): - return CLI_DEFAULT - raise SystemExit( - "librashader-cli not on PATH. Install or pass --cli. " - "Stock binary lives at " - r"C:\Users\Ed\scoop\apps\librashader\0.12.0\librashader-cli.exe" - ) - - -def parse_crop(spec: str) -> str | None: - """Parse a ``--crop`` value into an ffmpeg ``crop=...`` filter, or ``None`` to skip cropping.""" - if spec in ("none", "", "off"): - return None - if spec == "auto": - return "auto" - parts = [int(p) for p in spec.replace(" ", "").split(",")] - if len(parts) != 4: - raise SystemExit("--crop wants auto, none, or L,T,R,B") - left, top, right, bottom = parts - return f"crop=iw-{left + right}:ih-{top + bottom}:{left}:{top}" - - -def detect_crop(path: Path) -> str | None: - """Run ``ffmpeg cropdetect`` on ``path`` and return the auto-detected crop filter (or ``None``).""" - proc = subprocess.run( - [ - "ffmpeg", - "-hide_banner", - "-i", - str(path), - "-vf", - "cropdetect=limit=24:round=2:reset=0", - "-frames:v", - "1", - "-f", - "null", - "-", - ], - capture_output=True, - text=True, - ) - crop = None - for line in (proc.stderr or "").splitlines(): - if "crop=" in line: - crop = line.rsplit("crop=", 1)[-1].strip() - return f"crop={crop}" if crop else None - - -def run_ffmpeg(src: Path, dest: Path, vf: str) -> None: - """Run ``ffmpeg -y`` with the given ``-vf`` filter to render a single still ``src`` -> ``dest``.""" - dest.parent.mkdir(parents=True, exist_ok=True) - subprocess.check_call( - [ - "ffmpeg", - "-y", - "-hide_banner", - "-loglevel", - "error", - "-i", - str(src), - "-vf", - vf, - "-frames:v", - "1", - "-update", - "1", - str(dest), - ] - ) - - -def run_librashader( - cli: Path, - slang: Path, - src: Path, - dest: Path, - *, - runtime: str, - dimensions: str, - params: str, -) -> None: - """Render ``src`` through ``slang`` via the ``librashader-cli`` binary into ``dest``.""" - dest.parent.mkdir(parents=True, exist_ok=True) - dest.parent.mkdir(parents=True, exist_ok=True) - cmd = [ - str(cli), - "render", - "-r", - runtime, - "-d", - dimensions, - "--params", - params, - "-p", - str(slang), - "-i", - str(src), - "-o", - str(dest), - ] - print(" ".join(cmd)) - subprocess.check_call(cmd) - - -def main(argv: list[str] | None = None) -> int: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--in", dest="src", type=Path, required=True) - parser.add_argument("--out", dest="dest", type=Path, required=True) - parser.add_argument("--preset", choices=("auto", "lottes", "cyclon"), default="auto") - parser.add_argument( - "--nn", - type=int, - default=1, - help="nearest integer pre-scale (Affinity zoom). 1 = none, 3 = 300%%", - ) - parser.add_argument( - "--shader-scale", - default="100%", - help="librashader -d on the (possibly nn-scaled) image. 100%%, 300%%, 600%%, or WIDTHxHEIGHT", - ) - parser.add_argument("--runtime", default="d3d11") - parser.add_argument("--auto-max", type=int, default=720) - parser.add_argument("--crop", default="none", help="auto | none | L,T,R,B") - parser.add_argument("--cli", type=Path, default=None) - parser.add_argument("--dry-run", action="store_true") - parser.add_argument( - "--shaderglass", - action="store_true", - help="launch ShaderGlass with the matching .sgp (manual snapshot)", - ) - parser.add_argument( - "--nn-only", - action="store_true", - help="only do the neighbor pre-scale; do not call librashader", - ) - args = parser.parse_args(argv) - - if not args.src.is_file(): - raise SystemExit(f"missing {args.src}") - if args.nn < 1: - raise SystemExit("--nn must be >= 1") - - width, height = probe_size(args.src) - name, sgp, slang, params = pick_preset(width, height, args.preset, args.auto_max) - print(f"size {width}x{height} preset={name} nn={args.nn} shader-scale={args.shader_scale}") - print(f"sgp {sgp}") - print(f"slang {slang}") - - work = args.src - tmp_nn: Path | None = None - if args.nn > 1: - tmp_nn = args.dest.with_name(args.dest.stem + f"_nn{args.nn}.png") - vf = f"scale=iw*{args.nn}:ih*{args.nn}:flags=neighbor,format=rgba" - print("ffmpeg", "-i", work, "-vf", vf, tmp_nn) - if not args.dry_run: - run_ffmpeg(work, tmp_nn, vf) - work = tmp_nn - - if args.nn_only: - if tmp_nn is None: - raise SystemExit("--nn-only needs --nn > 1") - if not args.dry_run: - shutil.copy2(tmp_nn, args.dest) - print(f"wrote {args.dest} (nn only)") - return 0 - - cli = None if args.dry_run else find_cli(args.cli) - if args.dry_run: - print( - "librashader-cli render", - "-r", - args.runtime, - "-d", - args.shader_scale, - "--params", - params, - "-p", - slang, - "-i", - work, - "-o", - args.dest, - ) - else: - assert cli is not None - if not slang.is_file(): - raise SystemExit(f"missing slang {slang}") - run_librashader( - cli, - slang, - work, - args.dest, - runtime=args.runtime, - dimensions=args.shader_scale, - params=params, - ) - - crop_spec = parse_crop(args.crop) - if crop_spec == "auto" and not args.dry_run: - crop_filter = detect_crop(args.dest) - if crop_filter: - cropped = args.dest.with_name(args.dest.stem + "_crop.png") - run_ffmpeg(args.dest, cropped, crop_filter) - shutil.move(str(cropped), str(args.dest)) - elif crop_spec and crop_spec != "auto" and not args.dry_run: - cropped = args.dest.with_name(args.dest.stem + "_crop.png") - run_ffmpeg(args.dest, cropped, crop_spec) - shutil.move(str(cropped), str(args.dest)) - - print(f"wrote {args.dest}") - if args.shaderglass: - exe = shutil.which("shaderglass") or shutil.which("ShaderGlass") - if not exe: - raise SystemExit("shaderglass not on PATH") - subprocess.Popen([exe, str(sgp)]) - return 0 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/twitter/lift_catbox.py b/scripts/twitter/lift_catbox.py deleted file mode 100644 index fbc642c..0000000 --- a/scripts/twitter/lift_catbox.py +++ /dev/null @@ -1,20 +0,0 @@ -"""Deprecated. - -Bulk lift has been retired. Use `tw.py fallback --id --media-id --role ---confirm-origin-unavailable` after confirming the origin is unavailable. -""" -from __future__ import annotations - -import sys - - -def main(argv: list[str] | None = None) -> int: # pragma: no cover - deprecated entry - del argv - sys.stderr.write( - "lift_catbox.py is retired; use tw.py fallback with --confirm-origin-unavailable\n" - ) - return 2 - - -if __name__ == "__main__": - raise SystemExit(main()) diff --git a/scripts/twitter/unfinished_bloat/fetch_thread.py b/scripts/twitter/unfinished_bloat/fetch_thread.py deleted file mode 100644 index 606d677..0000000 --- a/scripts/twitter/unfinished_bloat/fetch_thread.py +++ /dev/null @@ -1,971 +0,0 @@ -"""Offline TweetDetail parser, pure cursor walker, and global -human-paced :class:`RequestGovernor` for ``twitter.fetch_thread``. - -This module owns three typed components: - -1. The typed boundary dataclasses (:class:`Cursor`, - :class:`TweetDetailPage`) and the pure JSON-to-typed-data - parser :func:`parse_tweet_detail_page`. The parser inspects - the TweetDetail wire path - ``data.threaded_conversation_with_injections_v2.instructions`` - and converts typed timeline entries into existing - :class:`twitter.models.PostData` instances. Cursors with - eligible kinds (``Bottom``, ``ShowMore``, ``ShowMoreThreads``) - are kept; Top and unknown cursors are dropped without - raising. Tweets wrapped in - :class:`TweetWithVisibilityResults` are unwrapped via their - inner ``tweet`` field. - -2. The pure FIFO continuation walker - :func:`walk_continuation_queue` and the pure post-group - merger :func:`merge_post_groups`. The walker has no - transport, no sleeper, no request cap, and no roles of its - own; the future orchestration wraps every provider call in - :meth:`RequestGovernor.issue`. - -3. The global human-paced :class:`RequestGovernor` (and its - :class:`RequestBudgetExceeded` budget exception). The - governor owns the ONLY global request count, the ONLY - pacing state, and the ONLY ordered ``roles`` log. One - governor per run wraps every provider call later - (bootstrap homepage, bootstrap JS, tip TweetDetail, root - TweetDetail, every continuation). - -This module deliberately does NOT contain HTTP code, cookie -loading, transaction helpers, the production transport, or -the CLI. Those live in later tasks of the same track and -reuse the parser, the walker, and the governor declared here. -""" -from __future__ import annotations - -import sys -from collections import deque -from collections.abc import Callable, Mapping, Sequence -from dataclasses import dataclass -from datetime import datetime -from pathlib import Path -from typing import Protocol - -# --- canonical package-or-script import shim --- -# Both `python -m twitter.fetch_thread` (package-mode) and -# `python scripts/twitter/fetch_thread.py` (script-mode) must work. -try: - from .models import PostData, PostMetrics -except ImportError: # pragma: no cover - script-mode import - sys.path.insert(0, str(Path(__file__).resolve().parent)) - from models import PostData, PostMetrics - - -# --- fetcher-local types --- - - -# --- provider-error taxonomy (Task 9) --- - - -class FetchError(RuntimeError): - """Base class for fetcher errors that fail closed. - - The fetcher never retries, never logs, and never echoes - response bodies, URLs, headers, cursor values, or secret - values in exception messages. Status code and exception - type are the only diagnostic surfaces. - """ - - -class RequestBudgetExceeded(FetchError): - """Raised when the global request budget is exhausted. - - Inherits from :class:`FetchError` so callers can catch the - full fetcher-error family with one ``except`` clause. The - class remains a :class:`RuntimeError` transitively because - :class:`FetchError` is a :class:`RuntimeError`. Existing - governor tests that assert the ``RuntimeError`` lineage - continue to pass. - """ - - -class AuthError(FetchError): - """HTTP 401 or 403. The authenticated session is no longer valid.""" - - -class RateLimitError(FetchError): - """HTTP 429. Twitter rate limit hit; no automatic retry.""" - - -class ProviderError(FetchError): - """Any non-2xx that is not 401/403/429, or a transport-layer - failure (``TimeoutError``, ``ConnectionError``, ``OSError``) - raised by the wrapped operation.""" - - -ELIGIBLE_CURSOR_TYPES: frozenset[str] = frozenset( - {"Bottom", "ShowMore", "ShowMoreThreads"}, -) - - -@dataclass(frozen=True) -class Cursor: - """A typed cursor entry from a TweetDetail response.""" - value: str - cursor_type: str - - def __post_init__(self) -> None: - if not self.value: - raise ValueError("Cursor.value must be non-empty") - if self.cursor_type not in ELIGIBLE_CURSOR_TYPES: - raise ValueError( - f"Cursor.cursor_type {self.cursor_type!r} is not in eligible set" - ) - - -@dataclass(frozen=True) -class TweetDetailPage: - """Parsed TweetDetail page: typed posts, eligible cursors, and - the focal post's ``legacy.conversation_id_str``.""" - posts: tuple[PostData, ...] - cursors: tuple[Cursor, ...] - focal_conversation_id: str | None = None - - -# --- typed raw-boundary helpers --- - - -def _from_wire_text(node: Mapping[str, object], key: str) -> str: - raw = node.get(key) - if not isinstance(raw, str): - raise KeyError(f"required string field {key!r}") - return raw - - -def _from_wire_int(node: Mapping[str, object], key: str) -> int: - raw = node.get(key) - if isinstance(raw, bool) or not isinstance(raw, int): - raise KeyError(f"required int field {key!r}") - return raw - - -def _from_wire_optional_int( - node: Mapping[str, object], key: str, -) -> int | None: - """Accept ``int`` or numeric-string. Anything else is rejected - with ``KeyError``. ``None`` propagates as ``None``.""" - raw = node.get(key) - if raw is None: - return None - if isinstance(raw, bool) or not isinstance(raw, int): - if isinstance(raw, str) and raw.lstrip("-").isdigit(): - return int(raw) - raise KeyError(f"optional int field {key!r} has wrong type") - return raw - - -# --- per-tweet normalization --- - - -def _find_item_content( - entry: Mapping[str, object], -) -> Mapping[str, object] | None: - """Return the entry's ``itemContent``-shaped payload, accepting - all three layouts seen in real TweetDetail wire responses: - ``entry.content.itemContent``, ``entry.item.itemContent``, - and ``entry.itemContent`` directly.""" - content = entry.get("content") - if isinstance(content, Mapping): - ic = content.get("itemContent") - if isinstance(ic, Mapping): - return ic - item = entry.get("item") - if isinstance(item, Mapping): - ic = item.get("itemContent") - if isinstance(ic, Mapping): - return ic - ic = entry.get("itemContent") - if isinstance(ic, Mapping): - return ic - return None - - -def _find_cursor_source(entry: Mapping[str, object]) -> Mapping[str, object] | None: - """Return the entry's cursor source mapping, accepting all - three layouts (top-level content, module item.itemContent, - direct itemContent).""" - content = entry.get("content") - if isinstance(content, Mapping) and "cursorType" in content: - return content - item = entry.get("item") - if isinstance(item, Mapping): - item_content = item.get("itemContent") - if ( - isinstance(item_content, Mapping) - and "cursorType" in item_content - ): - return item_content - ic = entry.get("itemContent") - if isinstance(ic, Mapping) and "cursorType" in ic: - return ic - return None - - -def _entry_type(entry: Mapping[str, object]) -> object: - content = entry.get("content") - if isinstance(content, Mapping): - return content.get("entryType") - return None - - -def _resolve_post_id(tweet: Mapping[str, object]) -> str: - rest_id = tweet.get("rest_id") - if isinstance(rest_id, str) and rest_id: - return rest_id - legacy = tweet.get("legacy") - if isinstance(legacy, Mapping): - legacy_id = legacy.get("id_str") - if isinstance(legacy_id, str) and legacy_id: - return legacy_id - raise KeyError("tweet: missing rest_id and legacy.id_str") - - -def _resolve_user(tweet: Mapping[str, object]) -> tuple[str, str]: - """Return ``(author, handle)`` for the tweet. - - Current shape first: ``user_results.result.core.name`` and - ``user_results.result.core.screen_name``. BOTH fields must be - non-empty strings for the current shape to win; otherwise - fall back to the legacy pair - ``user_results.result.legacy.name`` and - ``user_results.result.legacy.screen_name``. The legacy pair - is mandatory once the current shape is rejected. - """ - core = tweet.get("core") - user_results_node: object = None - if isinstance(core, Mapping): - user_results_node = core.get("user_results") - user: object = None - if isinstance(user_results_node, Mapping): - user = user_results_node.get("result") - if isinstance(user, Mapping): - # Current shape: user.core.name / user.core.screen_name. - # Wins only when BOTH fields are non-empty strings; - # otherwise fall through to the legacy pair. - user_core = user.get("core") - if isinstance(user_core, Mapping): - name_node = user_core.get("name") - screen_node = user_core.get("screen_name") - if ( - isinstance(name_node, str) - and name_node - and isinstance(screen_node, str) - and screen_node - ): - return str(name_node), str(screen_node) - # Fallback: user.legacy.name / user.legacy.screen_name - user_legacy = user.get("legacy") - if isinstance(user_legacy, Mapping): - name = _from_wire_text(user_legacy, "name") - handle = _from_wire_text(user_legacy, "screen_name") - return name, handle - raise KeyError("tweet: missing user_results.result core or legacy") - - -def _resolve_text(tweet: Mapping[str, object]) -> str: - note_tweet = tweet.get("note_tweet") - if isinstance(note_tweet, Mapping): - results = note_tweet.get("note_tweet_results") - if isinstance(results, Mapping): - inner = results.get("result") - if isinstance(inner, Mapping): - txt = inner.get("text") - if isinstance(txt, str) and txt: - return txt - legacy = tweet.get("legacy") - if isinstance(legacy, Mapping): - full_text = legacy.get("full_text") - if isinstance(full_text, str) and full_text: - return full_text - raise KeyError("tweet: missing note_tweet text and legacy.full_text") - - -def _normalize_created_at(created_at: str) -> str: - """Twitter ``created_at`` → ``YYYY-MM-DD HH:MM:SS``.""" - dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") - return dt.strftime("%Y-%m-%d %H:%M:%S") - - -def _resolve_media_urls(legacy: Mapping[str, object]) -> tuple[str, ...]: - """Collect media URLs from ``legacy.extended_entities.media[]`` - (falling back to ``entities.media[]``). Each photo URL is added; - each video contributes its highest-bitrate ``video/mp4`` - variant URL. Duplicates are dropped, preserving order.""" - ee_node = legacy.get("extended_entities") - ents: object = ee_node if isinstance(ee_node, Mapping) else None - if ents is None: - ents = legacy.get("entities") - if not isinstance(ents, Mapping): - return () - media_node = ents.get("media") - if not isinstance(media_node, list): - return () - urls: list[str] = [] - for m in media_node: - if not isinstance(m, Mapping): - continue - photo_url = m.get("media_url_https") - if not isinstance(photo_url, str): - photo_url = m.get("media_url") - if isinstance(photo_url, str) and photo_url: - urls.append(photo_url) - video_info = m.get("video_info") - if isinstance(video_info, Mapping): - variants_node = video_info.get("variants") - if isinstance(variants_node, list): - best: tuple[int, str] | None = None - for v in variants_node: - if not isinstance(v, Mapping): - continue - if v.get("content_type") != "video/mp4": - continue - vurl = v.get("url") - if not isinstance(vurl, str): - continue - br_node = v.get("bitrate") - if isinstance(br_node, bool) or not isinstance(br_node, int): - br = 0 - else: - br = br_node - if best is None or br > best[0]: - best = (br, vurl) - if best is not None: - urls.append(best[1]) - seen: set[str] = set() - deduped: list[str] = [] - for u in urls: - if u not in seen: - seen.add(u) - deduped.append(u) - return tuple(deduped) - - -def _build_post_data(tweet: Mapping[str, object]) -> PostData: - """Convert an unwrapped tweet mapping into a typed - :class:`PostData`. Raises ``KeyError`` or ``TypeError`` on - missing or wrong-typed required fields.""" - post_id = _resolve_post_id(tweet) - author, handle = _resolve_user(tweet) - text = _resolve_text(tweet) - legacy_node = tweet.get("legacy") - if not isinstance(legacy_node, Mapping): - raise KeyError(f"tweet {post_id}: missing legacy") - created_at = _from_wire_text(legacy_node, "created_at") - timestamp = _normalize_created_at(created_at) - reply_to_node = legacy_node.get("in_reply_to_status_id_str") - reply_to_id: str | None = ( - str(reply_to_node) - if isinstance(reply_to_node, str) and reply_to_node - else None - ) - quote_node = legacy_node.get("quoted_status_id_str") - quote_of_id: str | None = ( - str(quote_node) - if isinstance(quote_node, str) and quote_node - else None - ) - media_urls = _resolve_media_urls(legacy_node) - views_node = tweet.get("views") - views_count: int | None - if isinstance(views_node, Mapping): - views_count = _from_wire_optional_int(views_node, "count") - else: - views_count = None - metrics = PostMetrics( - reply_count=_from_wire_int(legacy_node, "reply_count"), - repost_count=_from_wire_int(legacy_node, "retweet_count"), - like_count=_from_wire_int(legacy_node, "favorite_count"), - view_count=views_count, - ) - return PostData( - post_id=post_id, - author=author, - handle=handle, - text=text, - timestamp=timestamp, - media_urls=media_urls, - reply_to_id=reply_to_id, - quote_of_id=quote_of_id, - metrics=metrics, - ) - - -def _normalize_tweet_entry(entry: Mapping[str, object]) -> PostData | None: - """Return a :class:`PostData` if ``entry`` carries a Tweet. - Returns ``None`` for cursors, modules, tombstone / unavailable - results, or entries whose inner ``__typename`` is not ``Tweet``.""" - entry_type = _entry_type(entry) - item_type = entry.get("itemType") - if ( - entry_type not in ("TimelineTimelineItem", "TimelineTweet") - and item_type not in ("TimelineTimelineItem", "TimelineTweet") - ): - return None - ic = _find_item_content(entry) - if ic is None: - return None - tweet_results = ic.get("tweet_results") - if not isinstance(tweet_results, Mapping): - return None - inner = tweet_results.get("result") - if not isinstance(inner, Mapping): - return None - if inner.get("__typename") == "TweetWithVisibilityResults": - inner = inner.get("tweet") - if not isinstance(inner, Mapping): - return None - if inner.get("__typename") != "Tweet": - return None - return _build_post_data(inner) - - -def _normalize_cursor_entry(entry: Mapping[str, object]) -> Cursor | None: - """Return a :class:`Cursor` for ``entry`` if it is a - TimelineTimelineCursor with an eligible ``cursorType``. - Top and unknown kinds are silently dropped.""" - cursor_source = _find_cursor_source(entry) - if cursor_source is None: - return None - entry_type = _entry_type(entry) - item_type = entry.get("itemType") - if ( - entry_type != "TimelineTimelineCursor" - and item_type != "TimelineTimelineCursor" - ): - return None - cursor_type_node = cursor_source.get("cursorType") - if not isinstance(cursor_type_node, str): - return None - if cursor_type_node not in ELIGIBLE_CURSOR_TYPES: - return None - value_node = cursor_source.get("value") - if isinstance(value_node, str): - value = value_node - elif isinstance(value_node, int) and not isinstance(value_node, bool): - value = str(value_node) - else: - return None - try: - return Cursor(value=value, cursor_type=cursor_type_node) - except ValueError: - return None - - -def _extract_focal_conversation_id( - entry: Mapping[str, object], focal_id: str, -) -> str | None: - """Return ``legacy.conversation_id_str`` from the wrapped tweet - whose id equals ``focal_id``. Inspects only the current entry; - iterators like :func:`_walk_entries` are responsible for - visiting every nested timeline node and calling this function - on each.""" - ic = _find_item_content(entry) - if ic is None: - return None - tweet_results = ic.get("tweet_results") - if not isinstance(tweet_results, Mapping): - return None - inner = tweet_results.get("result") - if not isinstance(inner, Mapping): - return None - if inner.get("__typename") == "TweetWithVisibilityResults": - inner = inner.get("tweet") - if not isinstance(inner, Mapping): - return None - cand_id = inner.get("rest_id") - if not isinstance(cand_id, str) or not cand_id: - legacy = inner.get("legacy") - if isinstance(legacy, Mapping): - cand_id = legacy.get("id_str") - if not isinstance(cand_id, str) or cand_id != focal_id: - return None - legacy = inner.get("legacy") - if not isinstance(legacy, Mapping): - return None - cid = legacy.get("conversation_id_str") - if isinstance(cid, str) and cid: - return cid - return None - - -def _walk_entries( - entries: Sequence[object], focal_id: str | None, -) -> tuple[list[PostData], list[Cursor], str | None]: - """Walk entries / modules iteratively via an explicit stack. - - The walk is bounded to the typed entries / modules tree. It - does NOT recursively scan tweet / quote / user payloads for - cursor-looking strings; cursors are only read from the - typed cursor source locations at the entry / module level. - - Source order is preserved: the stack is initialised with - the entries in reverse so the first ``pop`` yields the - first source entry, and module items are pushed in reverse - so the same invariant applies recursively. - - Returns ``(posts, cursors, focal_conversation_id_or_none)``. - """ - posts: list[PostData] = [] - cursors: list[Cursor] = [] - focal_conversation_id: str | None = None - stack: list[object] = list(reversed(list(entries))) - while stack: - node = stack.pop() - if not isinstance(node, Mapping): - continue - post = _normalize_tweet_entry(node) - if post is not None: - posts.append(post) - cursor = _normalize_cursor_entry(node) - if cursor is not None: - cursors.append(cursor) - if focal_id is not None and focal_conversation_id is None: - cid = _extract_focal_conversation_id(node, focal_id) - if cid is not None: - focal_conversation_id = cid - content = node.get("content") - if isinstance(content, Mapping) and content.get( - "entryType" - ) == "TimelineTimelineModule": - items_node = content.get("items") - if isinstance(items_node, list): - for item in reversed(items_node): - stack.append(item) - return posts, cursors, focal_conversation_id - - -def parse_tweet_detail_page( - raw: object, *, focal_id: str | None = None, -) -> TweetDetailPage: - """Parse a TweetDetail response into typed posts, eligible - cursors, and the focal post's ``legacy.conversation_id_str``. - - The wire path is - ``data.threaded_conversation_with_injections_v2.instructions``; - this function walks only those instructions and the module - contents they reference. It does not recursively scan - tweet / quote / user payloads for cursor-looking strings. - - Args: - raw: The deserialised wire JSON. Accepts ``object`` so - the caller can hand anything to the parser; a - non-mapping ``raw`` produces an empty page rather - than raising. - focal_id: When set, the parser walks all entries - (including those nested in TimelineTimelineModule - content.items[]) to find the entry whose id equals - ``focal_id`` and exposes its - ``legacy.conversation_id_str`` as - ``page.focal_conversation_id``. - - Returns: - A :class:`TweetDetailPage`. ``posts`` is deduped by id - with last-seen data preserved (first-seen order intact); - ``cursors`` is deduped by value with first-seen wins. - """ - if not isinstance(raw, Mapping): - return TweetDetailPage( - posts=(), cursors=(), focal_conversation_id=None, - ) - data_node = raw.get("data") - if not isinstance(data_node, Mapping): - return TweetDetailPage( - posts=(), cursors=(), focal_conversation_id=None, - ) - threaded = data_node.get("threaded_conversation_with_injections_v2") - if not isinstance(threaded, Mapping): - return TweetDetailPage( - posts=(), cursors=(), focal_conversation_id=None, - ) - instructions = threaded.get("instructions") - if not isinstance(instructions, list): - return TweetDetailPage( - posts=(), cursors=(), focal_conversation_id=None, - ) - all_posts: list[PostData] = [] - all_cursors: list[Cursor] = [] - final_focal: str | None = None - for instruction in instructions: - if not isinstance(instruction, Mapping): - continue - itype = instruction.get("type") - if itype == "TimelineAddEntries": - entries = instruction.get("entries") - if isinstance(entries, list): - posts, cursors, focal = _walk_entries(entries, focal_id) - all_posts.extend(posts) - all_cursors.extend(cursors) - if final_focal is None and focal is not None: - final_focal = focal - elif itype == "TimelineAddToModule": - module_items = instruction.get("moduleItems") - if isinstance(module_items, list): - posts, cursors, focal = _walk_entries(module_items, focal_id) - all_posts.extend(posts) - all_cursors.extend(cursors) - if final_focal is None and focal is not None: - final_focal = focal - elif itype == "TimelineReplaceEntry": - entry = instruction.get("entry") - if isinstance(entry, Mapping): - posts, cursors, focal = _walk_entries([entry], focal_id) - all_posts.extend(posts) - all_cursors.extend(cursors) - if final_focal is None and focal is not None: - final_focal = focal - by_id: dict[str, PostData] = {} - order: list[str] = [] - for post in all_posts: - if post.post_id not in by_id: - order.append(post.post_id) - by_id[post.post_id] = post - deduped_posts: list[PostData] = [by_id[oid] for oid in order] - seen_cursor_values: set[str] = set() - deduped_cursors: list[Cursor] = [] - for c in all_cursors: - if c.value not in seen_cursor_values: - seen_cursor_values.add(c.value) - deduped_cursors.append(c) - return TweetDetailPage( - posts=tuple(deduped_posts), - cursors=tuple(deduped_cursors), - focal_conversation_id=final_focal, - ) - - -# --- root resolution (Task 3) --- - - -def resolve_root_from_tip(tip: str, page: TweetDetailPage) -> str: - """Resolve the conversation root from a parsed tip page. - - Returns ``page.focal_conversation_id``, which - :func:`parse_tweet_detail_page` extracted from the focal tip - post's ``legacy.conversation_id_str``. - - Verifies that ``tip`` is present in ``page.posts`` and that - ``page.focal_conversation_id`` is a non-empty string. Neither - a parent-chain walk nor a ``reply_to_id is None`` heuristic is - used; the resolved root must come from the focal post's - conversation id. Parent-chain closure is validated later after - the tip+root page merge. - - Args: - tip: The post id of the input tip (the user's ``--tip`` - value). - page: A parsed :class:`TweetDetailPage` from - :func:`parse_tweet_detail_page`. - - Returns: - The resolved conversation root post id. - - Raises: - ValueError: ``tip`` is not present in ``page.posts``. - ValueError: ``page.focal_conversation_id`` is ``None`` or - an empty string. - """ - tip_ids = {post.post_id for post in page.posts} - if tip not in tip_ids: - raise ValueError(f"tip {tip!r} not present in page.posts") - focal = page.focal_conversation_id - if not focal: - raise ValueError( - "page.focal_conversation_id is missing or empty; " - "cannot resolve conversation root" - ) - return focal - - -# --- tip + root page merge (Task 4) --- - - -def merge_post_groups( - *groups: Sequence[PostData], -) -> tuple[PostData, ...]: - """Merge multiple post sequences into a deduped tuple. - - Last-seen data wins: when the same ``post_id`` appears in more - than one group, the later group's :class:`PostData` replaces the - earlier one. First-insertion order is preserved: the position of - each id in the output is the position of its first appearance in - the iteration order. Pure function; no I/O, network, or sample - constants. - """ - seen: dict[str, PostData] = {} - order: list[str] = [] - for group in groups: - for post in group: - if post.post_id not in seen: - order.append(post.post_id) - seen[post.post_id] = post - return tuple(seen[pid] for pid in order) - - -# --- continuation queue walker (Task 5) --- - - -@dataclass(frozen=True) -class ContinuationWalkResult: - """Outcome of :func:`walk_continuation_queue`. - - ``posts`` is the deduped union of the initial page's posts and - every continuation page's posts, with last-seen data wins and - first-insertion order preserved. ``fetched_cursor_count`` is - the number of times the walker invoked ``fetch_page`` (i.e., - the number of unseen eligible cursors it followed). - """ - posts: tuple[PostData, ...] - fetched_cursor_count: int - - -def walk_continuation_queue( - initial_page: TweetDetailPage, - *, - fetch_page: Callable[[Cursor], TweetDetailPage], -) -> ContinuationWalkResult: - """Walk the eligible explicit reply cursor queue, FIFO. - - Seeds the queue with ``initial_page.cursors``. For each unseen - cursor value (deduped by ``Cursor.value``), invokes - ``fetch_page`` exactly once, merges the returned page's posts - into the accumulator via :func:`merge_post_groups`, and - enqueues the returned page's own cursors. Iteration terminates - when the queue empties. - - The walker has no HTTP, sleep, retry, request cap, root/tip - special case, or logging. The future global ``RequestGovernor`` - wraps ``fetch_page`` and is the only request cap; depth / - loop behaviour is unbounded here. - - Exceptions raised by ``fetch_page`` propagate immediately on - the call that raised them. The walker does not catch, swallow, - or re-invoke the callback after a failure. - - Args: - initial_page: The starting :class:`TweetDetailPage`. Its - ``posts`` seed the accumulator; its ``cursors`` seed - the FIFO queue. - fetch_page: Callable invoked with one :class:`Cursor` and - returning the parsed page for that continuation. - - Returns: - A :class:`ContinuationWalkResult` with the merged posts - and the count of callback invocations. - """ - queue: deque[Cursor] = deque(initial_page.cursors) - seen_cursor_values: set[str] = set() - accumulator: tuple[PostData, ...] = initial_page.posts - fetched_cursor_count: int = 0 - while queue: - cursor = queue.popleft() - if cursor.value in seen_cursor_values: - continue - seen_cursor_values.add(cursor.value) - next_page = fetch_page(cursor) - fetched_cursor_count += 1 - accumulator = merge_post_groups(accumulator, next_page.posts) - for next_cursor in next_page.cursors: - if next_cursor.value not in seen_cursor_values: - queue.append(next_cursor) - return ContinuationWalkResult( - posts=accumulator, - fetched_cursor_count=fetched_cursor_count, - ) - - -# --- global human-paced request governor (Task 8) --- - - -class Sleeper(Protocol): - """The duck-typed surface the governor depends on. - - Production passes a real sleeper that blocks for - ``seconds``; tests pass ``RecordingSleeper`` from - :mod:`twitter.tests.fetch_thread_helpers`, which records - durations without blocking. - """ - - def sleep(self, seconds: float) -> None: - """Block (or record) for ``seconds`` seconds.""" - - -class RequestGovernor: - """Global, human-paced request envelope. - - Exactly one :class:`RequestGovernor` wraps every provider - call later in the orchestration (bootstrap homepage, - bootstrap JS, tip TweetDetail, root TweetDetail, every - continuation). The governor owns the **only** global request - count and the **only** pacing state. The pure cursor walker - :func:`walk_continuation_queue` carries no - ``max_requests``, no request index, no sleeper, and no role - log of its own. - - Behaviour: - - - ``max_requests >= 1`` and ``min_delay >= 0`` are enforced - at construction; non-empty ``role`` is enforced at every - ``issue``. - - Before every ``issue``: if ``count >= max_requests`` the - governor raises :class:`RequestBudgetExceeded` **before** - sleeping and **before** invoking the operation. No sleep - is appended, no role is appended. - - The very first attempted call does not sleep. Every later - attempted call sleeps exactly once with the configured - ``min_delay`` before invoking the operation. - - The role is appended and counted exactly once, - immediately before invoking the operation. If the - operation raises, the attempted call remains counted; the - governor does not catch, retry, or strip the role. - - No logging, no HTTP, no URL inspection, no cursor - inspection, no retry, no concurrency, no time polling, - no automatic backoff. - """ - - def __init__( - self, - *, - max_requests: int = 8, - min_delay: float = 5.0, - sleeper: Sleeper, - ) -> None: - if not isinstance(max_requests, int) or isinstance(max_requests, bool): - raise ValueError( - f"max_requests must be int, got {type(max_requests).__name__}", - ) - if max_requests < 1: - raise ValueError( - f"max_requests must be >= 1, got {max_requests}", - ) - if not isinstance(min_delay, (int, float)) or isinstance(min_delay, bool): - raise ValueError( - f"min_delay must be numeric, got {type(min_delay).__name__}", - ) - if float(min_delay) < 0.0: - raise ValueError( - f"min_delay must be >= 0, got {min_delay}", - ) - if sleeper is None: - raise ValueError("sleeper must not be None") - self._max_requests: int = max_requests - self._min_delay: float = float(min_delay) - self._sleeper: Sleeper = sleeper - self._roles: list[str] = [] - - @property - def count(self) -> int: - """The number of attempted calls counted so far.""" - return len(self._roles) - - @property - def roles(self) -> tuple[str, ...]: - """The ordered tuple of roles appended for attempted calls.""" - return tuple(self._roles) - - def issue( - self, - role: str, - operation: Callable[[], tuple[int, str]], - ) -> tuple[int, str]: - """Wrap ``operation`` in the pacing + budget envelope. - - Args: - role: A non-empty label for the attempted call - (e.g., ``"bootstrap_homepage"``, - ``"tip_tweet_detail"``, - ``"continuation_tweet_detail"``). Appended to - :attr:`roles` exactly once on attempted calls. - operation: A zero-argument callable returning - ``(status_code, body_text)``. The governor - invokes it at most once. - - Returns: - The ``(status_code, body_text)`` tuple returned by - ``operation``. - - Raises: - ValueError: ``role`` is empty or not a string. - RequestBudgetExceeded: ``count >= max_requests`` - before this call. Raised **before** sleeping and - **before** invoking ``operation``; the operation - is not called, no role is appended, no sleep is - recorded. - Exception: any exception raised by ``operation`` - propagates immediately. The attempted call is - counted (the role was already appended), and the - governor does not retry. - """ - if not isinstance(role, str): - raise ValueError( - f"role must be str, got {type(role).__name__}", - ) - if not role: - raise ValueError("role must be non-empty") - if self.count >= self._max_requests: - raise RequestBudgetExceeded( - f"request budget of {self._max_requests} exhausted " - f"before issuing role {role!r}", - ) - if self.count > 0: - self._sleeper.sleep(self._min_delay) - self._roles.append(role) - return operation() - - -# --- shared provider-error adapter (Task 9) --- - - -def issue_provider_request( - governor: RequestGovernor, - *, - role: str, - operation: Callable[[], tuple[int, str]], -) -> str: - """Run a single provider request through ``governor`` and - translate the outcome into either a body string or a typed - :class:`FetchError` subclass. - - Behaviour: - - - The wrapped ``operation`` is invoked exactly once through - :meth:`RequestGovernor.issue`. The governor owns the - pacing, the role log, and the budget cap. - - 2xx responses (``200``-``299``) return the body string - unchanged. - - ``401`` and ``403`` raise :class:`AuthError`. - - ``429`` raises :class:`RateLimitError`. - - Any other non-2xx status raises :class:`ProviderError`. - - ``TimeoutError``, ``ConnectionError``, and ``OSError`` raised - by ``operation`` are translated into :class:`ProviderError` - chained from the original via ``raise ... from exc``. - - ``RequestBudgetExceeded`` (raised by the governor when the - cap is reached) propagates unchanged. It is already a - :class:`FetchError` subclass and remains a - :class:`RuntimeError`. - - Arbitrary programmer/parser exceptions (e.g., - :class:`ValueError`) propagate unchanged; they are not - mislabelled as a provider failure. - - Diagnostic policy: exception messages are limited to the - numeric HTTP status and the exception class name. They never - include the response body, the URL, header values, cursor - values, or any secret value. The ``role`` argument is the - caller's own label and is intentionally not echoed into the - exception. - """ - try: - status, body = governor.issue(role, operation) - except (TimeoutError, ConnectionError, OSError) as exc: - raise ProviderError("transport error") from exc - if 200 <= status <= 299: - return body - if status in (401, 403): - raise AuthError(f"http {status}") - if status == 429: - raise RateLimitError(f"http {status}") - raise ProviderError(f"http {status}") diff --git a/scripts/twitter/unfinished_bloat/tests/fetch_thread_helpers.py b/scripts/twitter/unfinished_bloat/tests/fetch_thread_helpers.py deleted file mode 100644 index d539ae9..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fetch_thread_helpers.py +++ /dev/null @@ -1,206 +0,0 @@ -"""Test helpers for the Twitter full-conversation fetcher. - -Owns the canonical test-truth constants that production -``twitter.fetch_thread`` mirrors (the equality is asserted at -import time once production exists). Also owns the offline fakes -used by every test that exercises the fetcher without touching -the network: ``FakeTransport``, ``RecordingSleeper``, and -``FakeTransaction``. - -This module deliberately stays a *test-only* module: it must not -import ``twitter.fetch_thread`` (production does not exist yet -when the helpers are first created) and must not read cookies, -open sockets, or sleep for real. -""" -from __future__ import annotations - -import json -from dataclasses import dataclass, field -from pathlib import Path - -# --- canonical test-truth constants (production mirrors these) --- - - -KNOWN_ROOT_ID: str = "1650678968255913985" -KNOWN_TIP_ID: str = "1651282559287042048" - -EXPECTED_16: frozenset[str] = frozenset({ - "1650678968255913985", # kenpex OP - "1650685805776732160", # abductee_org -> OP - "1650915682136240129", # streleav -> OP - "1651056827839180800", # wadetb -> OP - "1651086935467917312", # ChristerEricson -> OP - "1651168030557077504", # SebAaltonen -> OP - "1651253961524142081", # kenpex -> wadetb - "1651254727160795137", # kenpex -> ChristerEricson - "1651268028795961344", # NOTimothyLottes -> kenpex - "1651282559287042048", # wvo (the input tip) - "1651295755293036544", # ErikHaliewicz -> OP - "1651508233243095043", # npatsiouras -> kenpex Christer branch - "1651510424649744385", # ChristerEricson -> npatsiouras - "1651511316010663936", # ChristerEricson -> self - "1651579576089300992", # JaceCear -> OP - "1651636988267888641", # kenpex -> JaceCear -}) - -# --- fixture loader --- - - -FIXTURES_DIR: Path = ( - Path(__file__).resolve().parent / "fixtures" / "fetch_thread" -) - - -def load_fixture(name: str) -> str: - """Return the raw UTF-8 text of a sanitized fixture by basename. - - The caller may pass ``"tip_tweet_detail"`` or - ``"tip_tweet_detail.json"``; either form resolves to the same - file under :data:`FIXTURES_DIR`. - """ - filename = name if name.endswith(".json") else f"{name}.json" - return (FIXTURES_DIR / filename).read_text(encoding="utf-8") - - -def load_json_fixture(name: str) -> object: - """Parse the JSON fixture with ``name`` (with or without the - ``.json`` suffix) and return the deserialised value. - - Return type is ``object`` because the wire shape varies: pages, - cursor lists, error envelopes, and homepage HTML fragments all - round-trip through here. - """ - return json.loads(load_fixture(name)) - - -# --- fakes --- - - -@dataclass -class RecordingSleeper: - """Records each requested sleep duration without blocking. - - The fetcher's :class:`RequestGovernor` calls ``sleep(min_delay)`` - after the first request of every role; tests assert the exact - sequence and value via ``self.durations``. - """ - - durations: list[float] = field(default_factory=list) - - def sleep(self, seconds: float) -> None: - """Append the requested duration; return immediately.""" - self.durations.append(seconds) - - -@dataclass -class FakeTransport: - """In-memory transport. Issues no HTTP requests. - - The fetcher's production transport implements - ``fetch(role, url) -> (status, body)``. The fake mirrors that - signature so production and tests agree on the call surface. - - Attributes: - responses: Role-keyed canned responses. Each value is a - ``(status_code, body_text)`` pair. Missing keys raise - ``AssertionError`` so tests fail loudly when a new role - is introduced without a corresponding fixture. - calls: Per-``fetch`` call record. Each entry is a dict with - ``"role"`` and ``"url"`` keys, in call order. - """ - - responses: dict[str, tuple[int, str]] = field(default_factory=dict) - calls: list[dict[str, str]] = field(default_factory=list) - _transaction: object | None = None - - def set_transaction(self, transaction: object) -> None: - """Store a fake transaction for later inspection. - - The real ``RequestsSessionTransport`` calls this once the - two bootstrap requests succeed. Tests pass a - :class:`FakeTransaction` (or any object) and assert the - order by reading ``self._transaction`` later. - """ - self._transaction = transaction - - def fetch(self, role: str, url: str) -> tuple[int, str]: - """Return the canned response for ``role`` and record the call. - - Raises ``AssertionError`` if ``role`` is not keyed in - :attr:`responses`; this is the offline equivalent of a 5xx - when the live provider returns nothing for an unknown role. - """ - if role not in self.responses: - raise AssertionError( - f"FakeTransport: no response registered for role {role!r}" - ) - self.calls.append({"role": role, "url": url}) - return self.responses[role] - - -class FakeTransaction: - """Fake MIT ``XClientTransaction``-shaped object for tests. - - The real constructor signature (verified offline) is - ``ClientTransaction(home_page_response, ondemand_file_response, - random_keyword=None, random_number=None)``. Tests capture both - arguments verbatim so they can assert the fetcher passed the - parsed ``BeautifulSoup`` and raw JS body to the factory in the - right order. The ``generate_transaction_id`` method is directly - defined on this class (delegating to the module-level - :func:`generate_transaction_id` helper) so the fake exactly - matches the real call surface. - """ - - def __init__( - self, - home_soup: object, - js_body: object, - **kwargs: object, - ) -> None: - self.home_soup = home_soup - self.js_body = js_body - self.kwargs = kwargs - - def generate_transaction_id( - self, - method: str, - path: str, - **kwargs: object, - ) -> str: - """Return a fixed test transaction id.""" - return generate_transaction_id(method, path, **kwargs) - - -_FIXED_TRANSACTION_ID: str = "fixed-test-transaction-id" - - -def generate_transaction_id( - method: str, - path: str, - **kwargs: object, -) -> str: - """Return a fixed transaction id for offline tests. - - The id value is irrelevant to behaviour; tests assert that the - production code calls ``generate_transaction_id(method, path, - **kwargs)`` and that the result reaches the request header. - """ - return _FIXED_TRANSACTION_ID - - -def make_fake_transaction( - home_soup: object, - js_body: object, - **kwargs: object, -) -> FakeTransaction: - """Construct a :class:`FakeTransaction` with the production - call signature so a test's ``transaction_factory`` matches the - real ``ClientTransaction(home_page_response, ondemand_file_response)`` - two-positional-arg constructor. - - Exposed as a free function (rather than baking the signature - into ``FakeTransaction``) so tests can pass it as a - ``transaction_factory`` callable to the fetcher. - """ - return FakeTransaction(home_soup, js_body, **kwargs) diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/.gitkeep b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_1.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_1.json deleted file mode 100644 index 3872092..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_1.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-1-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000001", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000001", - "core": { - "name": "fixture-depth-1-author", - "screen_name": "fixture-depth-1-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000001", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:01:00 +0000 2026", - "full_text": "fixture-depth-1-text", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-1-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMore", - "value": "fixture-depth-2" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_2.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_2.json deleted file mode 100644 index 3715a8d..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_2.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-2-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000002", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000002", - "core": { - "name": "fixture-depth-2-author", - "screen_name": "fixture-depth-2-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000002", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:02:00 +0000 2026", - "full_text": "fixture-depth-2-text", - "in_reply_to_status_id_str": "1700000000000000001", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-2-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "Bottom", - "value": "fixture-depth-3" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_3.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_3.json deleted file mode 100644 index 668fd40..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_3.json +++ /dev/null @@ -1,66 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-3-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000003", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000003", - "core": { - "name": "fixture-depth-3-author", - "screen_name": "fixture-depth-3-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000003", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:03:00 +0000 2026", - "full_text": "fixture-depth-3-text", - "in_reply_to_status_id_str": "1700000000000000002", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-3-cursor-echo-current", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "Bottom", - "value": "fixture-depth-3" - } - }, - { - "entryId": "depth-3-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMoreThreads", - "value": "fixture-depth-4" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_4.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_4.json deleted file mode 100644 index 4171f62..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_4.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-4-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000004", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000004", - "core": { - "name": "fixture-depth-4-author", - "screen_name": "fixture-depth-4-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000004", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:04:00 +0000 2026", - "full_text": "fixture-depth-4-text", - "in_reply_to_status_id_str": "1700000000000000003", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-4-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMore", - "value": "fixture-depth-5" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_5.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_5.json deleted file mode 100644 index 250a9e6..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_5.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-5-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000005", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000005", - "core": { - "name": "fixture-depth-5-author", - "screen_name": "fixture-depth-5-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000005", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:05:00 +0000 2026", - "full_text": "fixture-depth-5-text", - "in_reply_to_status_id_str": "1700000000000000004", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-5-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "Bottom", - "value": "fixture-depth-6" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_6.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_6.json deleted file mode 100644 index 255f2e4..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_6.json +++ /dev/null @@ -1,58 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-6-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000006", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000006", - "core": { - "name": "fixture-depth-6-author", - "screen_name": "fixture-depth-6-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000006", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:06:00 +0000 2026", - "full_text": "fixture-depth-6-text", - "in_reply_to_status_id_str": "1700000000000000005", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "depth-6-cursor-forward", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMoreThreads", - "value": "fixture-depth-7" - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_7.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_7.json deleted file mode 100644 index 3e997ae..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_depth_7.json +++ /dev/null @@ -1,50 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "depth-7-tweet", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1700000000000000007", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "8100000007", - "core": { - "name": "fixture-depth-7-author", - "screen_name": "fixture-depth-7-handle" - } - } - } - }, - "legacy": { - "id_str": "1700000000000000007", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:07:00 +0000 2026", - "full_text": "fixture-depth-7-text", - "in_reply_to_status_id_str": "1700000000000000006", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_tip_branch.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_tip_branch.json deleted file mode 100644 index e009e52..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/continuation_tip_branch.json +++ /dev/null @@ -1,89 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "continuation-tweet-lottes", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651268028795961344", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "444444444", - "core": { - "name": "fixture-author-lottes-continuation", - "screen_name": "fixture-handle-lottes-continuation" - } - } - } - }, - "legacy": { - "id_str": "1651268028795961344", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:03:00 +0000 2026", - "full_text": "fixture-text-lottes-continuation", - "in_reply_to_status_id_str": "1651253961524142081", - "in_reply_to_user_id_str": "333333333", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "continuation-tweet-tip", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651282559287042048", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "555555555", - "core": { - "name": "fixture-author-tip-continuation", - "screen_name": "fixture-handle-tip-continuation" - } - } - } - }, - "legacy": { - "id_str": "1651282559287042048", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:04:00 +0000 2026", - "full_text": "fixture-text-tip-continuation", - "in_reply_to_status_id_str": "1651268028795961344", - "in_reply_to_user_id_str": "444444444", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/root_tweet_detail.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/root_tweet_detail.json deleted file mode 100644 index ad9247a..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/root_tweet_detail.json +++ /dev/null @@ -1,539 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "tweet-root", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1650678968255913985", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "955913985", - "core": { - "name": "fixture-author-root-page", - "screen_name": "fixture-handle-root-page" - } - } - } - }, - "legacy": { - "id_str": "1650678968255913985", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-root-page", - "in_reply_to_status_id_str": null, - "quoted_status_id_str": null, - "reply_count": 5, - "retweet_count": 6, - "favorite_count": 7 - }, - "views": { - "count": 1000 - } - } - } - } - } - } - ] - }, - { - "type": "TimelineAddToModule", - "moduleItems": [ - { - "entryId": "mod-01-abductee_org-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1650685805776732160", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "976732160", - "core": { - "name": "fixture-author-abductee_org-page", - "screen_name": "fixture-handle-abductee-org-page" - } - } - } - }, - "legacy": { - "id_str": "1650685805776732160", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-abductee-org-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-02-streleav-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1650915682136240129", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "936240129", - "core": { - "name": "fixture-author-streleav-page", - "screen_name": "fixture-handle-streleav-page" - } - } - } - }, - "legacy": { - "id_str": "1650915682136240129", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-streleav-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-03-wadetb-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651056827839180800", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "939180800", - "core": { - "name": "fixture-author-wadetb-page", - "screen_name": "fixture-handle-wadetb-page" - } - } - } - }, - "legacy": { - "id_str": "1651056827839180800", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-wadetb-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-04-Christer-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651086935467917312", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "967917312", - "core": { - "name": "fixture-author-Christer-page", - "screen_name": "fixture-handle-Christer-page" - } - } - } - }, - "legacy": { - "id_str": "1651086935467917312", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Christer-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-05-Seb-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651168030557077504", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "957077504", - "core": { - "name": "fixture-author-Seb-page", - "screen_name": "fixture-handle-Seb-page" - } - } - } - }, - "legacy": { - "id_str": "1651168030557077504", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Seb-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-06-kenpex-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651253961524142081", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "924142081", - "core": { - "name": "fixture-author-kenpex-page", - "screen_name": "fixture-handle-kenpex-page" - } - } - } - }, - "legacy": { - "id_str": "1651253961524142081", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-kenpex-to-wadetb-page", - "in_reply_to_status_id_str": "1651056827839180800", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-07-kenpex-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651254727160795137", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "960795137", - "core": { - "name": "fixture-author-kenpex-page", - "screen_name": "fixture-handle-kenpex-page-2" - } - } - } - }, - "legacy": { - "id_str": "1651254727160795137", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-kenpex-to-Christer-page", - "in_reply_to_status_id_str": "1651086935467917312", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-08-Erik-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651295755293036544", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "993036544", - "core": { - "name": "fixture-author-Erik-page", - "screen_name": "fixture-handle-Erik-page" - } - } - } - }, - "legacy": { - "id_str": "1651295755293036544", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Erik-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-09-npatsiouras-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651508233243095043", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "943095043", - "core": { - "name": "fixture-author-npatsiouras-page", - "screen_name": "fixture-handle-npatsiouras-page" - } - } - } - }, - "legacy": { - "id_str": "1651508233243095043", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-npatsiouras-page", - "in_reply_to_status_id_str": "1651254727160795137", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-10-Christer-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651510424649744385", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "949744385", - "core": { - "name": "fixture-author-Christer-page", - "screen_name": "fixture-handle-Christer-page-2" - } - } - } - }, - "legacy": { - "id_str": "1651510424649744385", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Christer-to-npatsiouras-page", - "in_reply_to_status_id_str": "1651508233243095043", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-11-Christer-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651511316010663936", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "910663936", - "core": { - "name": "fixture-author-Christer-page", - "screen_name": "fixture-handle-Christer-page-3" - } - } - } - }, - "legacy": { - "id_str": "1651511316010663936", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Christer-self-page", - "in_reply_to_status_id_str": "1651510424649744385", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-12-Jace-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651579576089300992", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "989300992", - "core": { - "name": "fixture-author-Jace-page", - "screen_name": "fixture-handle-Jace-page" - } - } - } - }, - "legacy": { - "id_str": "1651579576089300992", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-Jace-page", - "in_reply_to_status_id_str": "1650678968255913985", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-13-kenpex-page", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651636988267888641", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "967888641", - "core": { - "name": "fixture-author-kenpex-page", - "screen_name": "fixture-handle-kenpex-page-3" - } - } - } - }, - "legacy": { - "id_str": "1651636988267888641", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-kenpex-to-Jace-page", - "in_reply_to_status_id_str": "1651579576089300992", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/tip_tweet_detail.json b/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/tip_tweet_detail.json deleted file mode 100644 index 1afde0f..0000000 --- a/scripts/twitter/unfinished_bloat/tests/fixtures/fetch_thread/tip_tweet_detail.json +++ /dev/null @@ -1,277 +0,0 @@ -{ - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": [ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "tweet-root", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1650678968255913985", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "111111111", - "core": { - "name": "fixture-author-root", - "screen_name": "fixture-handle-root" - }, - "legacy": { - "name": "fixture-author-root-legacy-should-be-ignored", - "screen_name": "fixture-handle-root-legacy-should-be-ignored" - } - } - } - }, - "legacy": { - "id_str": "1650678968255913985", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:00:00 +0000 2026", - "full_text": "fixture-text-root-legacy", - "in_reply_to_status_id_str": null, - "in_reply_to_user_id_str": null, - "quoted_status_id_str": null, - "reply_count": 1, - "retweet_count": 2, - "favorite_count": 3 - }, - "views": {"count": 100} - } - } - } - } - }, - { - "entryId": "conversationthread-1", - "content": { - "entryType": "TimelineTimelineModule", - "items": [ - { - "entryId": "mod-tweet-wadetb", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651056827839180800", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "222222222", - "core": { - "name": "fixture-author-wadetb", - "screen_name": "fixture-handle-wadetb" - } - } - } - }, - "legacy": { - "id_str": "1651056827839180800", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:01:00 +0000 2026", - "full_text": "fixture-text-wadetb-full", - "in_reply_to_status_id_str": "1650678968255913985", - "in_reply_to_user_id_str": "111111111", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0, - "extended_entities": { - "media": [ - { - "media_url_https": "https://pbs.twimg.com/media/fixture-photo-wadetb.jpg", - "type": "photo" - } - ] - } - } - } - } - } - } - }, - { - "entryId": "mod-tweet-kenpex", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651253961524142081", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "333333333", - "core": { - "name": "fixture-author-kenpex", - "screen_name": "fixture-handle-kenpex" - } - } - } - }, - "note_tweet": { - "note_tweet_results": { - "result": { - "text": "fixture-note-tweet-text-kenpex-override" - } - } - }, - "legacy": { - "id_str": "1651253961524142081", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:02:00 +0000 2026", - "full_text": "fixture-text-kenpex-legacy-should-be-ignored", - "in_reply_to_status_id_str": "1651056827839180800", - "in_reply_to_user_id_str": "222222222", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - } - } - } - } - } - }, - { - "entryId": "mod-tweet-lottes", - "itemType": "TimelineTimelineItem", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "TweetWithVisibilityResults", - "tweet": { - "__typename": "Tweet", - "rest_id": "1651268028795961344", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "444444444", - "core": { - "name": "fixture-author-lottes", - "screen_name": "fixture-handle-lottes" - } - } - } - }, - "legacy": { - "id_str": "1651268028795961344", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:03:00 +0000 2026", - "full_text": "fixture-text-lottes", - "in_reply_to_status_id_str": "1651253961524142081", - "in_reply_to_user_id_str": "333333333", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0, - "extended_entities": { - "media": [ - { - "media_url_https": "https://pbs.twimg.com/media/fixture-photo-lottes-poster.jpg", - "type": "video", - "video_info": { - "variants": [ - { - "content_type": "video/mp4", - "url": "https://video.twimg.com/fixture-video-low.mp4", - "bitrate": 320000 - }, - { - "content_type": "video/mp4", - "url": "https://video.twimg.com/fixture-video-med.mp4", - "bitrate": 832000 - }, - { - "content_type": "video/mp4", - "url": "https://video.twimg.com/fixture-video-high.mp4", - "bitrate": 1280000 - }, - { - "content_type": "application/x-mpegURL", - "url": "https://video.twimg.com/fixture-video.m3u8", - "bitrate": 0 - } - ] - } - } - ] - } - } - } - } - } - } - } - }, - { - "entryId": "mod-tweet-tip", - "itemType": "TimelineTweet", - "item": { - "itemContent": { - "tweet_results": { - "result": { - "__typename": "Tweet", - "rest_id": "1651282559287042048", - "core": { - "user_results": { - "result": { - "__typename": "User", - "id_str": "555555555", - "core": { - "name": "fixture-author-tip", - "screen_name": "fixture-handle-tip" - } - } - } - }, - "legacy": { - "id_str": "1651282559287042048", - "conversation_id_str": "1650678968255913985", - "created_at": "Mon Apr 26 12:04:00 +0000 2026", - "full_text": "fixture-text-tip", - "in_reply_to_status_id_str": "1651268028795961344", - "in_reply_to_user_id_str": "444444444", - "quoted_status_id_str": null, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0 - }, - "views": {"count": "1234"} - } - } - } - } - }, - { - "entryId": "mod-cursor-top", - "itemType": "TimelineTimelineCursor", - "item": { - "itemContent": { - "cursorType": "Top", - "value": "fixture-top-cursor-token-should-be-ignored" - } - } - } - ] - } - } - ] - } - ] - } - } -} diff --git a/scripts/twitter/unfinished_bloat/tests/test_fetch_thread.py b/scripts/twitter/unfinished_bloat/tests/test_fetch_thread.py deleted file mode 100644 index 82285fe..0000000 --- a/scripts/twitter/unfinished_bloat/tests/test_fetch_thread.py +++ /dev/null @@ -1,2025 +0,0 @@ -"""Offline tests for ``twitter.fetch_thread``. Zero network calls. - -Task 2 boundary tests: parse a sanitized ``tip_tweet_detail.json`` -fixture into the fetcher-local :class:`TweetDetailPage` whose -``posts`` are the existing :class:`models.PostData` instances. -Asserts exactly five posts on the tip spine, the resolved -``focal_conversation_id`` is the known root, note-tweet text wins -over ``legacy.full_text``, current-shape user core maps cleanly, -visibility-wrapped tweets unwrap, photo URLs land in -``media_urls``, the highest-bitrate MP4 variant is selected, and -numeric-string view counts parse to ``int``. - -The :class:`Cursor` and :class:`TweetDetailPage` types are -fetcher-local; only those two new dataclasses appear in this -task. The :class:`Cursor`\s ``__post_init__`` rejects ineligible -cursor kinds at construction time; the parser never reaches -construction for Top or unknown kinds. - -Task 8 tests: :class:`RequestGovernor` is the global, human-paced -request envelope that wraps every provider call. The walker stays -pure (no pacing, no cap, no roles of its own). The governor owns -the only request count, the only pacing, and the only ordered -``roles`` log. Tests use the actual ``RecordingSleeper`` from -Task 1; no clock is involved (the governor does not poll time). -""" -from __future__ import annotations - -import unittest - -from twitter.fetch_thread import ( - AuthError, - Cursor, - FetchError, - ProviderError, - RateLimitError, - RequestBudgetExceeded, - RequestGovernor, - Sleeper, - TweetDetailPage, - issue_provider_request, - parse_tweet_detail_page, - resolve_root_from_tip, - walk_continuation_queue, -) -from twitter.models import PostData, PostMetrics -from twitter.tests.fetch_thread_helpers import ( - EXPECTED_16, - KNOWN_ROOT_ID, - KNOWN_TIP_ID, - RecordingSleeper, - load_json_fixture, -) - -TIP_ID = KNOWN_TIP_ID -ROOT_ID = KNOWN_ROOT_ID -WADETB_ID = "1651056827839180800" -KENPEX_ID = "1651253961524142081" -LOTTES_ID = "1651268028795961344" - - -class TipPageParserTests(unittest.TestCase): - """The fixture exercises one TimelineAddEntries instruction - containing one top-level TimelineTimelineItem (root) and one - TimelineTimelineModule (``content.items[]``) wrapping four - payload items plus one Top cursor. Parser must produce 5 - posts, 0 cursors, focal_id == root.""" - - def setUp(self) -> None: - self.raw = load_json_fixture("tip_tweet_detail") - - def test_parse_returns_exactly_five_post_data_with_exact_id_set(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - self.assertIsInstance(page, TweetDetailPage) - self.assertEqual(len(page.posts), 5) - ids = {p.post_id for p in page.posts} - self.assertEqual( - ids, - {ROOT_ID, WADETB_ID, KENPEX_ID, LOTTES_ID, TIP_ID}, - ) - # Each post is an existing models.PostData instance with - # an existing models.PostMetrics metrics field; no - # fetcher-local duplicates are introduced. - for post in page.posts: - self.assertIsInstance(post, PostData) - self.assertIsInstance(post.metrics, PostMetrics) - - def test_parent_chain_walks_root_to_tip_in_order(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - by_id = {p.post_id: p for p in page.posts} - expected_chain = [ - (TIP_ID, LOTTES_ID), - (LOTTES_ID, KENPEX_ID), - (KENPEX_ID, WADETB_ID), - (WADETB_ID, ROOT_ID), - (ROOT_ID, None), - ] - for post_id, parent_id in expected_chain: - self.assertEqual( - by_id[post_id].reply_to_id, - parent_id, - msg=f"post {post_id}: parent mismatch", - ) - - def test_focal_conversation_id_resolves_to_root(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - self.assertEqual(page.focal_conversation_id, ROOT_ID) - - def test_tip_page_has_zero_eligible_cursors(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - self.assertEqual(page.cursors, ()) - self.assertEqual(len(page.cursors), 0) - - def test_note_tweet_text_overrides_legacy_full_text(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - kenpex = next(p for p in page.posts if p.post_id == KENPEX_ID) - self.assertEqual( - kenpex.text, - "fixture-note-tweet-text-kenpex-override", - ) - - def test_current_core_user_mapping(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - root = next(p for p in page.posts if p.post_id == ROOT_ID) - self.assertEqual(root.author, "fixture-author-root") - self.assertEqual(root.handle, "fixture-handle-root") - # The legacy string on user_results.result is deliberately - # different in the fixture; the current core shape must - # win and the legacy fallback must NOT leak through. - self.assertNotIn("legacy-should-be-ignored", root.author) - self.assertNotIn("legacy-should-be-ignored", root.handle) - tip = next(p for p in page.posts if p.post_id == TIP_ID) - self.assertEqual(tip.author, "fixture-author-tip") - self.assertEqual(tip.handle, "fixture-handle-tip") - - def test_visibility_wrapped_lottes_post_is_parsed(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - lottes = next(p for p in page.posts if p.post_id == LOTTES_ID) - self.assertEqual(lottes.post_id, LOTTES_ID) - # visibility wrapper must be unwrapped; text reaches the - # parser through the inner Tweet. - self.assertEqual(lottes.text, "fixture-text-lottes") - - def test_photo_url_lands_in_media_urls(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - wadetb = next(p for p in page.posts if p.post_id == WADETB_ID) - self.assertIn( - "https://pbs.twimg.com/media/fixture-photo-wadetb.jpg", - wadetb.media_urls, - ) - - def test_highest_bitrate_mp4_selected_lower_bitrate_excluded(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - lottes = next(p for p in page.posts if p.post_id == LOTTES_ID) - # Highest bitrate (1280000) wins; lower bitrates are not - # appended. The HLS m3u8 variant is not video/mp4 so it - # is filtered out at the variant-content-type check. - self.assertIn( - "https://video.twimg.com/fixture-video-high.mp4", - lottes.media_urls, - ) - self.assertNotIn( - "https://video.twimg.com/fixture-video-low.mp4", - lottes.media_urls, - ) - self.assertNotIn( - "https://video.twimg.com/fixture-video-med.mp4", - lottes.media_urls, - ) - self.assertNotIn( - "https://video.twimg.com/fixture-video.m3u8", - lottes.media_urls, - ) - - def test_numeric_string_view_count_parses_as_int(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=TIP_ID) - tip = next(p for p in page.posts if p.post_id == TIP_ID) - self.assertEqual(tip.metrics.view_count, 1234) - self.assertIsInstance(tip.metrics.view_count, int) - - -class CursorConstructorTests(unittest.TestCase): - """The fetcher-local :class:`Cursor` rejects empty values and - ineligible kinds at construction. Top and unknown cursor types - never reach construction during parse (the parser returns - ``None`` for ineligible kinds).""" - - def test_eligible_cursor_constructs(self) -> None: - for kind in ("Bottom", "ShowMore", "ShowMoreThreads"): - with self.subTest(kind=kind): - cursor = Cursor(value="abcdef", cursor_type=kind) - self.assertEqual(cursor.value, "abcdef") - self.assertEqual(cursor.cursor_type, kind) - - def test_empty_value_raises(self) -> None: - with self.assertRaises(ValueError): - Cursor(value="", cursor_type="Bottom") - - def test_unknown_kind_raises(self) -> None: - with self.assertRaises(ValueError): - Cursor(value="abcdef", cursor_type="Top") - - -class ExpectedSixteenTests(unittest.TestCase): - """The test-vs-production invariant from the Task 1 helpers - module is mirrored here. The five-post tip spine is a subset - of the planned 16-post :data:`EXPECTED_16` set; this test - pins the overlap so the helpers and parser agree.""" - - def test_tip_spine_is_subset_of_expected_16(self) -> None: - page = parse_tweet_detail_page( - load_json_fixture("tip_tweet_detail"), focal_id=TIP_ID, - ) - page_ids = {p.post_id for p in page.posts} - self.assertTrue(page_ids.issubset(EXPECTED_16)) - - -# --- helpers for synthetic wire objects (in-memory, no fixture files) --- - - -def _wrap_tweet_result(tweet_inner: dict) -> dict: - """Wrap a single tweet ``result`` mapping as the body of a - TimelineTimelineItem entry's ``tweet_results.result``.""" - return { - "tweet_results": {"result": tweet_inner}, - } - - -def _tweet_result_with_user( - *, - rest_id: str, - author_legacy: str, - handle_legacy: str, - user_core: dict | None, - full_text: str, - created_at: str = "Mon Apr 26 12:00:00 +0000 2026", - conversation_id: str = "8888000000000000000", - in_reply_to: str | None = None, -) -> dict: - """Build a typed ``result`` mapping for one tweet (no wrapper). - - ``user_core`` is the dict placed on ``user_results.result.core`` - (or ``None`` to omit the current shape entirely). Legacy name / - screen_name on ``user_results.result.legacy`` are always present - so the production fallback path stays defined. - """ - user: dict = { - "__typename": "User", - "id_str": "777777777", - "legacy": { - "name": author_legacy, - "screen_name": handle_legacy, - }, - } - if user_core is not None: - user["core"] = user_core - return { - "__typename": "Tweet", - "rest_id": rest_id, - "core": {"user_results": {"result": user}}, - "legacy": { - "id_str": rest_id, - "conversation_id_str": conversation_id, - "created_at": created_at, - "full_text": full_text, - "in_reply_to_status_id_str": in_reply_to, - "quoted_status_id_str": None, - "reply_count": 0, - "retweet_count": 0, - "favorite_count": 0, - }, - } - - -def _page(instructions: list) -> object: - """Wrap a list of instructions in the data -> threaded path envelope.""" - return { - "data": { - "threaded_conversation_with_injections_v2": { - "instructions": instructions, - }, - }, - } - - -class CurrentShapeUserEdgeCasesTests(unittest.TestCase): - """Item 3: current-shape ``user.core.name`` / ``user.core.screen_name`` - must both be non-empty strings for the current shape to win. A - partial current shape (name only, screen_name missing, or empty - string screen_name) must fall back to ``user.legacy.name`` / - ``user.legacy.screen_name``. - - In-memory wire objects only; no fixture file is touched. - """ - - def test_current_shape_with_only_name_falls_back_to_legacy(self) -> None: - # Current core has name but no screen_name key at all. - # Legacy carries both name and screen_name. The parser - # MUST use the legacy pair (not return an empty handle). - tweet = _tweet_result_with_user( - rest_id="9999000000000000001", - author_legacy="fixture-legacy-edge-author", - handle_legacy="fixture-legacy-edge-handle", - user_core={"name": "fixture-edge-author"}, # no screen_name - full_text="fixture-edge-text-1", - ) - wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "entry-edge-1", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(tweet), - }, - }, - ], - }, - ]) - page = parse_tweet_detail_page(wire) - self.assertEqual(len(page.posts), 1) - post = page.posts[0] - self.assertEqual(post.author, "fixture-legacy-edge-author") - self.assertEqual(post.handle, "fixture-legacy-edge-handle") - - def test_current_shape_with_empty_screen_name_falls_back_to_legacy(self) -> None: - # Current core has name and an empty-string screen_name; - # an empty string is NOT a valid handle and must trigger - # the legacy fallback. - tweet = _tweet_result_with_user( - rest_id="9999000000000000002", - author_legacy="fixture-legacy-edge-author-2", - handle_legacy="fixture-legacy-edge-handle-2", - user_core={"name": "fixture-edge-author-2", "screen_name": ""}, - full_text="fixture-edge-text-2", - ) - wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "entry-edge-2", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(tweet), - }, - }, - ], - }, - ]) - page = parse_tweet_detail_page(wire) - self.assertEqual(len(page.posts), 1) - post = page.posts[0] - self.assertEqual(post.author, "fixture-legacy-edge-author-2") - self.assertEqual(post.handle, "fixture-legacy-edge-handle-2") - - -class DedupeCharacterizationTests(unittest.TestCase): - """Item 4: characterization tests for the documented dedupe - behaviour. These pin existing production behaviour rather than - force a RED. Pinning them as named tests guards against - regressions in dedupe order, value-key, and module-wrapped - cursor parsing. - """ - - def test_post_dedupe_last_seen_data_one_output(self) -> None: - # Two TimelineAddEntries entries share the same rest_id. - # The second entry carries a different author and text. - # After dedupe the output must contain exactly one post - # whose data comes from the LAST occurrence. - first = _tweet_result_with_user( - rest_id="1650678968255913985", - author_legacy="fixture-dedupe-handle-first", - handle_legacy="fixture-dedupe-handle-first", - user_core={"name": "fixture-dedupe-first", "screen_name": "fixture-dedupe-first"}, - full_text="text-dedupe-first", - created_at="Mon Apr 26 12:00:00 +0000 2026", - ) - second = _tweet_result_with_user( - rest_id="1650678968255913985", - author_legacy="fixture-dedupe-handle-second", - handle_legacy="fixture-dedupe-handle-second", - user_core={"name": "fixture-dedupe-second", "screen_name": "fixture-dedupe-second"}, - full_text="text-dedupe-second", - created_at="Mon Apr 26 12:05:00 +0000 2026", - ) - wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "tweet-dedupe-first", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(first), - }, - }, - { - "entryId": "tweet-dedupe-second", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(second), - }, - }, - ], - }, - ]) - page = parse_tweet_detail_page(wire) - self.assertEqual(len(page.posts), 1) - post = page.posts[0] - self.assertEqual(post.post_id, "1650678968255913985") - # Last-seen data wins: author is "fixture-dedupe-second" - # and text is "text-dedupe-second". - self.assertEqual(post.author, "fixture-dedupe-second") - self.assertEqual(post.text, "text-dedupe-second") - - def test_cursor_dedupe_by_value(self) -> None: - # Two TimelineTimelineCursor entries with the same value - # and the same eligible kind: only one Cursor must appear. - wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "cursor-a", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMore", - "value": "fixture-dedupe-cursor-value", - }, - }, - { - "entryId": "cursor-b", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMore", - "value": "fixture-dedupe-cursor-value", - }, - }, - ], - }, - ]) - page = parse_tweet_detail_page(wire) - self.assertEqual(len(page.cursors), 1) - self.assertEqual( - page.cursors[0].value, "fixture-dedupe-cursor-value", - ) - self.assertEqual(page.cursors[0].cursor_type, "ShowMore") - - def test_eligible_cursor_wrapped_under_item_itemcontent_parses(self) -> None: - # The cursor lives inside a TimelineTimelineModule's - # content.items[] entry whose payload is wrapped under - # item.itemContent (the "module" layout described in the - # brief). The parser must surface the eligible cursor. - wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "module-with-eligible-cursor", - "content": { - "entryType": "TimelineTimelineModule", - "items": [ - { - "entryId": "mod-cursor-showmore", - "itemType": "TimelineTimelineCursor", - "item": { - "itemContent": { - "cursorType": "ShowMore", - "value": "fixture-showmore-cursor-under-item", - }, - }, - }, - ], - }, - }, - ], - }, - ]) - page = parse_tweet_detail_page(wire) - self.assertEqual(len(page.cursors), 1) - cursor = page.cursors[0] - self.assertEqual( - cursor.value, "fixture-showmore-cursor-under-item", - ) - self.assertEqual(cursor.cursor_type, "ShowMore") - - -class ResolveRootFromTipTests(unittest.TestCase): - """Task 3: ``resolve_root_from_tip`` reads the resolved root - directly off ``page.focal_conversation_id`` (parsed from the - focal tip post's ``legacy.conversation_id_str``). - - The function is a pure page-level helper: it does NOT walk - parents, query the transport, or substitute a - ``reply_to_id is None`` heuristic. Parent-chain closure is - validated later after the tip+root page merge. - """ - - def test_returns_focal_conversation_id_for_known_tip(self) -> None: - # The existing tip_tweet_detail.json fixture has the - # focal tip post whose legacy.conversation_id_str equals - # the resolved root (1650678968255913985). Parsing the - # fixture with focal_id=KNOWN_TIP_ID gives a page whose - # focal_conversation_id is that root. The function must - # return it. - raw = load_json_fixture("tip_tweet_detail") - page = parse_tweet_detail_page(raw, focal_id=KNOWN_TIP_ID) - root = resolve_root_from_tip(KNOWN_TIP_ID, page) - self.assertEqual(root, KNOWN_ROOT_ID) - self.assertEqual(root, "1650678968255913985") - - def test_raises_when_tip_absent_from_page_posts(self) -> None: - # The tip is not in page.posts (the parsed page holds - # the five-post tip spine). An unknown tip must raise - # ValueError; the function does not silently fall back - # to a heuristic. - raw = load_json_fixture("tip_tweet_detail") - page = parse_tweet_detail_page(raw, focal_id=KNOWN_TIP_ID) - with self.assertRaises(ValueError): - resolve_root_from_tip("0000000000000000000", page) - - def test_raises_when_focal_conversation_id_is_none(self) -> None: - # The tip is in page.posts, but focal_conversation_id - # is None. The function must raise ValueError — the - # resolved root MUST come from the focal post's - # conversation_id_str, not from a parent-walk or - # reply_to_id is None heuristic. - tip_post = PostData( - post_id=KNOWN_TIP_ID, - author="fixture-author-tip", - handle="fixture-handle-tip", - text="fixture-text-tip", - timestamp="2026-08-24 00:00:00", - media_urls=(), - reply_to_id="1651268028795961344", - quote_of_id=None, - metrics=PostMetrics(0, 0, 0, None), - ) - page = TweetDetailPage( - posts=(tip_post,), - cursors=(), - focal_conversation_id=None, - ) - with self.assertRaises(ValueError): - resolve_root_from_tip(KNOWN_TIP_ID, page) - - def test_raises_when_focal_conversation_id_is_empty_string(self) -> None: - # An empty-string focal_conversation_id is the same - # kind of absent-value case as None and must also raise. - tip_post = PostData( - post_id=KNOWN_TIP_ID, - author="fixture-author-tip", - handle="fixture-handle-tip", - text="fixture-text-tip", - timestamp="2026-08-24 00:00:00", - media_urls=(), - reply_to_id="1651268028795961344", - quote_of_id=None, - metrics=PostMetrics(0, 0, 0, None), - ) - page = TweetDetailPage( - posts=(tip_post,), - cursors=(), - focal_conversation_id="", - ) - with self.assertRaises(ValueError): - resolve_root_from_tip(KNOWN_TIP_ID, page) - - def test_prefers_focal_conversation_id_over_reply_to_none_heuristic( - self, - ) -> None: - # Construct a page with two posts: - # - a "misleading" post whose reply_to_id is None - # (would be picked by a "first post with no parent" - # heuristic as the root) — id 9999000000000000000 - # - the tip itself, whose parent chain does NOT - # reach the misleading post - # The focal_conversation_id is the canonical root - # 1650678968255913985, which differs from the misleading - # id. The function MUST return the focal_conversation_id - # value, proving the heuristic is not in play. - misleading_root = PostData( - post_id="9999000000000000000", - author="fixture-misleading-author", - handle="fixture-misleading-handle", - text="fixture-misleading-text", - timestamp="2026-08-24 00:00:00", - media_urls=(), - reply_to_id=None, - quote_of_id=None, - metrics=PostMetrics(0, 0, 0, None), - ) - tip_post = PostData( - post_id=KNOWN_TIP_ID, - author="fixture-author-tip", - handle="fixture-handle-tip", - text="fixture-text-tip", - timestamp="2026-08-24 00:05:00", - media_urls=(), - reply_to_id="1651268028795961344", - quote_of_id=None, - metrics=PostMetrics(0, 0, 0, None), - ) - page = TweetDetailPage( - posts=(misleading_root, tip_post), - cursors=(), - focal_conversation_id=KNOWN_ROOT_ID, - ) - root = resolve_root_from_tip(KNOWN_TIP_ID, page) - # MUST be the focal_conversation_id value, NOT the - # misleading "no parent" post's id. - self.assertEqual(root, KNOWN_ROOT_ID) - self.assertNotEqual(root, misleading_root.post_id) - - -class RootPageParserTests(unittest.TestCase): - """Task 4: the root fixture exercises ``TimelineAddToModule`` - (a parser path not central in the tip fixture). It contains the - 14 root-side public ids from NOTE §2.1 (all of :data:`EXPECTED_16` - except the wvo tip ``1651282559287042048`` and the lottes post - ``1651268028795961344``). The tip fixture supplies those two. - No eligible cursors. The focal root post's - ``legacy.conversation_id_str`` is the root id. - """ - - def setUp(self) -> None: - self.raw = load_json_fixture("root_tweet_detail") - - def test_parse_returns_exactly_fourteen_post_data_with_exact_id_set( - self, - ) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=ROOT_ID) - self.assertIsInstance(page, TweetDetailPage) - self.assertEqual(len(page.posts), 14) - ids = {p.post_id for p in page.posts} - expected_root_only = set(EXPECTED_16) - {LOTTES_ID, TIP_ID} - self.assertEqual(ids, expected_root_only) - # Lottes and tip absent from root page. - self.assertNotIn(LOTTES_ID, ids) - self.assertNotIn(TIP_ID, ids) - # Each post is an existing models.PostData instance with - # an existing models.PostMetrics metrics field. - for post in page.posts: - self.assertIsInstance(post, PostData) - self.assertIsInstance(post.metrics, PostMetrics) - - def test_root_focal_conversation_id_is_root(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=ROOT_ID) - self.assertEqual(page.focal_conversation_id, ROOT_ID) - - def test_root_page_has_zero_eligible_cursors(self) -> None: - page = parse_tweet_detail_page(self.raw, focal_id=ROOT_ID) - self.assertEqual(page.cursors, ()) - self.assertEqual(len(page.cursors), 0) - - def test_parent_graph_uses_only_in_fixture_post_ids(self) -> None: - # Every non-root post in the root fixture must have its - # reply_to_id point at another post in the same fixture. - # (The lone lottes child of kenpex->wadetb lives in the - # tip page, not here, so the root page alone cannot satisfy - # lottes's closure; we only assert intra-fixture closure.) - page = parse_tweet_detail_page(self.raw, focal_id=ROOT_ID) - ids = {p.post_id for p in page.posts} - for post in page.posts: - if post.reply_to_id is None: - self.assertEqual(post.post_id, ROOT_ID) - continue - self.assertIn( - post.reply_to_id, ids, - msg=f"post {post.post_id}: parent {post.reply_to_id} absent from root fixture", - ) - - -class MergePostGroupsTests(unittest.TestCase): - """Task 4: ``merge_post_groups`` is the pure merger that joins - the tip and root pages into the final 16-post set without a - second query per post. - - - Last-seen data wins for posts appearing in multiple groups. - - First-insertion order is preserved: the first occurrence of - each id defines its position in the output. - - Returns a tuple of existing :class:`PostData` instances - (no copies, no synthetic PostData wrappers). - """ - - @staticmethod - def _parsed_pages() -> tuple: - tip_page = parse_tweet_detail_page( - load_json_fixture("tip_tweet_detail"), focal_id=TIP_ID, - ) - root_page = parse_tweet_detail_page( - load_json_fixture("root_tweet_detail"), focal_id=ROOT_ID, - ) - return tip_page, root_page - - def test_merge_tip_and_root_yields_expected_16(self) -> None: - from twitter.fetch_thread import merge_post_groups - tip_page, root_page = self._parsed_pages() - merged = merge_post_groups(tip_page.posts, root_page.posts) - ids = {p.post_id for p in merged} - self.assertEqual(ids, set(EXPECTED_16)) - self.assertEqual(len(merged), 16) - - def test_merge_one_output_per_id_no_duplicates(self) -> None: - from twitter.fetch_thread import merge_post_groups - tip_page, root_page = self._parsed_pages() - merged = merge_post_groups(tip_page.posts, root_page.posts) - ids = [p.post_id for p in merged] - self.assertEqual(len(ids), len(set(ids))) - - def test_last_seen_group_wins_on_overlap(self) -> None: - # Two PostData instances with the same post_id but - # different text/metrics. The second group wins. - from twitter.fetch_thread import merge_post_groups - first = PostData( - post_id="1650678968255913985", - author="fixture-dedupe-first", - handle="fixture-dedupe-first", - text="text-dedupe-first", - timestamp="2026-08-24 00:00:00", - media_urls=(), - reply_to_id=None, - quote_of_id=None, - metrics=PostMetrics(0, 0, 0, None), - ) - second = PostData( - post_id="1650678968255913985", - author="fixture-dedupe-second", - handle="fixture-dedupe-second", - text="text-dedupe-second", - timestamp="2026-08-24 00:05:00", - media_urls=(), - reply_to_id=None, - quote_of_id=None, - metrics=PostMetrics(1, 2, 3, 4), - ) - merged = merge_post_groups((first,), (second,)) - self.assertEqual(len(merged), 1) - post = merged[0] - self.assertEqual(post.text, "text-dedupe-second") - self.assertEqual(post.author, "fixture-dedupe-second") - self.assertEqual(post.metrics.like_count, 3) - self.assertEqual(post.metrics.reply_count, 1) - self.assertEqual(post.metrics.view_count, 4) - - def test_first_insertion_order_preserved(self) -> None: - # Tip posts appear first (5 ids, in tip order); root posts - # are appended after in root order. Overlap ids (root, - # wadetb, kenpex->wadetb) keep the tip's first-insertion - # position. - from twitter.fetch_thread import merge_post_groups - tip_page, root_page = self._parsed_pages() - merged = merge_post_groups(tip_page.posts, root_page.posts) - ids = [p.post_id for p in merged] - self.assertEqual( - ids, - [ - "1650678968255913985", - "1651056827839180800", - "1651253961524142081", - "1651268028795961344", - "1651282559287042048", - "1650685805776732160", - "1650915682136240129", - "1651086935467917312", - "1651168030557077504", - "1651254727160795137", - "1651295755293036544", - "1651508233243095043", - "1651510424649744385", - "1651511316010663936", - "1651579576089300992", - "1651636988267888641", - ], - ) - - def test_merged_parent_graph_closure(self) -> None: - # Every non-root post's reply_to_id must point at a post - # present in the merged set. The root itself may have no - # parent. - from twitter.fetch_thread import merge_post_groups - tip_page, root_page = self._parsed_pages() - merged = merge_post_groups(tip_page.posts, root_page.posts) - merged_ids = {p.post_id for p in merged} - for post in merged: - if post.post_id == ROOT_ID: - continue - self.assertIn( - post.reply_to_id, merged_ids, - msg=f"post {post.post_id}: parent {post.reply_to_id} missing from merged set", - ) - # The tip's parent chain must reach root through the - # captured set: tip -> lottes -> kenpex->wadetb -> wadetb -> root. - tip_post = next(p for p in merged if p.post_id == TIP_ID) - self.assertEqual(tip_post.reply_to_id, LOTTES_ID) - lottes = next(p for p in merged if p.post_id == LOTTES_ID) - self.assertEqual(lottes.reply_to_id, "1651253961524142081") - kenpex_to_wadetb = next(p for p in merged if p.post_id == "1651253961524142081") - self.assertEqual(kenpex_to_wadetb.reply_to_id, "1651056827839180800") - wadetb = next(p for p in merged if p.post_id == "1651056827839180800") - self.assertEqual(wadetb.reply_to_id, ROOT_ID) - - def test_merged_objects_remain_existing_post_data(self) -> None: - from twitter.fetch_thread import merge_post_groups - tip_page, root_page = self._parsed_pages() - merged = merge_post_groups(tip_page.posts, root_page.posts) - for post in merged: - self.assertIsInstance(post, PostData) - self.assertIsInstance(post.metrics, PostMetrics) - - -class ContinuationWalkerTests(unittest.TestCase): - """Task 5: ``walk_continuation_queue`` is the pure FIFO walker - that follows eligible explicit reply cursors off an initial - :class:`TweetDetailPage`. It carries no transport, sleeper, or - request cap: tests inject a callback that returns parsed pages - directly. - - The walker: - - - treats the initial page's ``cursors`` as the FIFO seed, - - calls the callback exactly once per *unseen* cursor value, - - merges returned posts via ``merge_post_groups`` - (last-seen wins, first-insertion order preserved), - - enqueues unseen eligible cursors from each returned page, - - stops when the FIFO is empty, - - re-raises callback exceptions immediately (no swallow, no retry). - - The fixture exercises one synthetic - ``ShowMoreThreads`` cursor that resolves to the tip branch - (lottes ``1651268028795961344`` and tip - ``1651282559287042048``). Combined with the root page's - 14 posts, the merged result must equal ``EXPECTED_16``. - """ - - CONTINUATION_CURSOR_VALUE: str = "fixture-show-more-tip-branch" - LOTTES_ID: str = "1651268028795961344" - TIP_ID: str = TIP_ID - - def _initial_page_with_synthetic_cursor(self) -> TweetDetailPage: - """Parse the root fixture and append one synthetic - ``ShowMoreThreads`` cursor. The walker treats this as the - only eligible continuation off the root page.""" - root_raw = load_json_fixture("root_tweet_detail") - root_page = parse_tweet_detail_page(root_raw, focal_id=ROOT_ID) - synthetic_cursor = Cursor( - value=self.CONTINUATION_CURSOR_VALUE, - cursor_type="ShowMoreThreads", - ) - return TweetDetailPage( - posts=root_page.posts, - cursors=(synthetic_cursor,), - focal_conversation_id=root_page.focal_conversation_id, - ) - - def _continuation_page(self) -> TweetDetailPage: - """Parse the sanitized tip-branch continuation fixture.""" - return parse_tweet_detail_page( - load_json_fixture("continuation_tip_branch"), - focal_id=ROOT_ID, - ) - - def test_walker_calls_callback_once_and_merges_to_expected_16(self) -> None: - # The walker imports a missing function name; the RED is - # the ImportError when ``walk_continuation_queue`` is not - # yet exported from twitter.fetch_thread. - from twitter.fetch_thread import walk_continuation_queue - - initial_page = self._initial_page_with_synthetic_cursor() - synthetic_cursor = initial_page.cursors[0] - # Pre-parse the continuation page so the callback and - # the verification lookup share the same parsed object - # instances (PostData is frozen; identity == "same object - # from one parse"). - continuation_page = self._continuation_page() - - captured: list[Cursor] = [] - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - captured.append(cursor) - self.assertEqual(cursor, synthetic_cursor) - self.assertEqual(cursor.value, self.CONTINUATION_CURSOR_VALUE) - self.assertEqual(cursor.cursor_type, "ShowMoreThreads") - return continuation_page - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - # Exactly one callback call: the synthetic cursor and - # nothing else (the continuation page exposes no eligible - # cursors, so the FIFO empties immediately after the merge). - self.assertEqual(len(captured), 1) - self.assertEqual(result.fetched_cursor_count, 1) - - # Merged post set equals EXPECTED_16 exactly: root14 + - # continuation2 (lottes + tip) without duplicates. - ids = {p.post_id for p in result.posts} - self.assertEqual(ids, set(EXPECTED_16)) - self.assertEqual(len(result.posts), 16) - self.assertIn(LOTTES_ID, ids) - self.assertIn(TIP_ID, ids) - - # One output per id: no duplicates in the merged tuple. - merged_ids = [p.post_id for p in result.posts] - self.assertEqual(len(merged_ids), len(set(merged_ids))) - - # Parent links: lottes -> kenpex->wadetb; tip -> lottes. - # Both links must be intact after the merge. - by_id = {p.post_id: p for p in result.posts} - self.assertEqual( - by_id[LOTTES_ID].reply_to_id, KENPEX_ID, - msg="lottes reply_to_id must be kenpex->wadetb", - ) - self.assertEqual( - by_id[TIP_ID].reply_to_id, LOTTES_ID, - msg="tip reply_to_id must be lottes", - ) - - # PostData instances preserved: shared ids (root, wadetb, - # kenpex->wadetb) come from the *initial* page (same - # object), not from copies or rewrites. Continuation-only - # ids (lottes, tip) come from the continuation page. - initial_by_id = {p.post_id: p for p in initial_page.posts} - continuation_by_id = { - p.post_id: p for p in continuation_page.posts - } - for shared_id in (ROOT_ID, WADETB_ID, KENPEX_ID): - self.assertIs( - by_id[shared_id], initial_by_id[shared_id], - msg=f"shared id {shared_id}: PostData instance must be preserved", - ) - for continuation_only_id in (LOTTES_ID, TIP_ID): - self.assertIs( - by_id[continuation_only_id], - continuation_by_id[continuation_only_id], - msg=( - f"continuation-only id {continuation_only_id}: " - f"PostData instance must come from continuation page" - ), - ) - - # Every post is the existing models.PostData type with the - # existing PostMetrics field intact. - for post in result.posts: - self.assertIsInstance(post, PostData) - self.assertIsInstance(post.metrics, PostMetrics) - - def test_callback_exception_propagates_immediately_no_second_call( - self, - ) -> None: - from twitter.fetch_thread import walk_continuation_queue - - class CallbackError(Exception): - pass - - initial_page = self._initial_page_with_synthetic_cursor() - call_count = 0 - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - nonlocal call_count - call_count += 1 - raise CallbackError( - f"simulated callback failure for {cursor.value}", - ) - - with self.assertRaises(CallbackError): - walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - # Exactly one call: the first (and only) cursor in the - # FIFO queue. The walker must not swallow the exception - # and must not re-invoke the callback before raising. - self.assertEqual(call_count, 1) - - def test_duplicate_initial_cursor_values_result_in_one_callback( - self, - ) -> None: - # Characterization: two distinct Cursor instances that - # share a ``value`` must collapse to one callback call. - # ``seen_cursor_values`` is the dedupe key, not the Cursor - # dataclass identity. - from twitter.fetch_thread import walk_continuation_queue - - root_raw = load_json_fixture("root_tweet_detail") - root_page = parse_tweet_detail_page(root_raw, focal_id=ROOT_ID) - duplicate_cursor = Cursor( - value=self.CONTINUATION_CURSOR_VALUE, - cursor_type="ShowMoreThreads", - ) - initial_page = TweetDetailPage( - posts=root_page.posts, - cursors=(duplicate_cursor, duplicate_cursor), - focal_conversation_id=root_page.focal_conversation_id, - ) - - call_count = 0 - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - nonlocal call_count - call_count += 1 - return self._continuation_page() - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - self.assertEqual(call_count, 1) - self.assertEqual(result.fetched_cursor_count, 1) - ids = {p.post_id for p in result.posts} - self.assertEqual(ids, set(EXPECTED_16)) - - -class FullDepthContinuationCharacterizationTests(unittest.TestCase): - """Task 6 characterization: the pure FIFO walker must follow - explicit continuation cursors through SEVEN reply depths with - no shallow depth limit, dedupe seen cursors, terminate naturally - on a leaf, and short-circuit when the initial page has no - cursors. - - The walker is the existing ``walk_continuation_queue`` in - ``twitter.fetch_thread``. These tests pin its behavior under - seven levels of nested reply continuation, with no production - change. The fixtures are seven sanitized raw TweetDetail - continuation pages (``continuation_depth_1.json`` through - ``continuation_depth_7.json``); depth-3 additionally echoes its - own/current cursor alongside the next cursor to prove - seen-cursor loop suppression. - """ - - DEPTH_POST_IDS: tuple[str, ...] = ( - "1700000000000000001", - "1700000000000000002", - "1700000000000000003", - "1700000000000000004", - "1700000000000000005", - "1700000000000000006", - "1700000000000000007", - ) - - def _initial_page_with_root_and_one_cursor( - self, - ) -> TweetDetailPage: - """Build an initial page with the known root post and - exactly one eligible ``ShowMoreThreads`` cursor valued - ``fixture-depth-1``. The seven sanitized depth fixtures - are loaded separately. - """ - root_tweet = _tweet_result_with_user( - rest_id=ROOT_ID, - author_legacy="fixture-root-author", - handle_legacy="fixture-root-handle", - user_core={ - "name": "fixture-root-author", - "screen_name": "fixture-root-handle", - }, - full_text="fixture-root-text", - in_reply_to=None, - ) - initial_wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "initial-root", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(root_tweet), - }, - }, - { - "entryId": "initial-cursor-depth-1", - "content": { - "entryType": "TimelineTimelineCursor", - "cursorType": "ShowMoreThreads", - "value": "fixture-depth-1", - }, - }, - ], - }, - ]) - return parse_tweet_detail_page(initial_wire, focal_id=ROOT_ID) - - def _parse_all_depth_pages( - self, - ) -> dict[str, TweetDetailPage]: - """Parse all seven sanitized raw TweetDetail continuation - pages and key them by the cursor value that fetches them: - page N is fetched via ``fixture-depth-N``. - """ - pages: dict[str, TweetDetailPage] = {} - for n in range(1, 8): - page = parse_tweet_detail_page( - load_json_fixture(f"continuation_depth_{n}"), - focal_id=ROOT_ID, - ) - pages[f"fixture-depth-{n}"] = page - return pages - - def test_initial_page_has_root_and_exactly_one_cursor(self) -> None: - # The initial page must contain the known root post and - # exactly one eligible ShowMoreThreads cursor valued - # ``fixture-depth-1``. This is the seed the walker consumes. - initial_page = self._initial_page_with_root_and_one_cursor() - self.assertEqual(len(initial_page.posts), 1) - self.assertEqual(initial_page.posts[0].post_id, ROOT_ID) - self.assertEqual(len(initial_page.cursors), 1) - cursor = initial_page.cursors[0] - self.assertEqual(cursor.value, "fixture-depth-1") - self.assertEqual(cursor.cursor_type, "ShowMoreThreads") - - def test_seven_levels_walked_callback_order_and_exact_parent_chain( - self, - ) -> None: - # The walker follows seven explicit continuation cursors - # FIFO to a leaf, with no depth limit. Result: exactly - # seven callbacks, in cursor order; eight posts in the - # accumulator (root + seven synthetic); parent chain - # walks back to the known root. - initial_page = self._initial_page_with_root_and_one_cursor() - pages_by_cursor = self._parse_all_depth_pages() - callback_calls: list[str] = [] - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - callback_calls.append(cursor.value) - return pages_by_cursor[cursor.value] - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - self.assertEqual(result.fetched_cursor_count, 7) - self.assertEqual( - callback_calls, - [ - "fixture-depth-1", - "fixture-depth-2", - "fixture-depth-3", - "fixture-depth-4", - "fixture-depth-5", - "fixture-depth-6", - "fixture-depth-7", - ], - ) - - ids = {p.post_id for p in result.posts} - self.assertEqual(ids, {ROOT_ID, *self.DEPTH_POST_IDS}) - self.assertEqual(len(result.posts), 8) - - by_id = {p.post_id: p for p in result.posts} - for n in range(6, 0, -1): - self.assertEqual( - by_id[self.DEPTH_POST_IDS[n]].reply_to_id, - self.DEPTH_POST_IDS[n - 1], - msg=f"depth-{n + 1} must reply to depth-{n}", - ) - self.assertEqual( - by_id[self.DEPTH_POST_IDS[0]].reply_to_id, - ROOT_ID, - msg="depth-1 must reply to the known root", - ) - self.assertIsNone(by_id[ROOT_ID].reply_to_id) - - for post in result.posts: - self.assertIsInstance(post, PostData) - self.assertIsInstance(post.metrics, PostMetrics) - - def test_repeated_current_cursor_is_not_called_twice_no_loop( - self, - ) -> None: - # Depth-3's fixture echoes its own incoming cursor - # (``fixture-depth-3``) alongside the next cursor - # (``fixture-depth-4``). The walker must skip the echo via - # seen-cursor dedupe; the callback is invoked exactly once - # for that value and the queue does not loop. - initial_page = self._initial_page_with_root_and_one_cursor() - pages_by_cursor = self._parse_all_depth_pages() - depth_3_call_count: int = 0 - callback_calls: list[str] = [] - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - nonlocal depth_3_call_count - callback_calls.append(cursor.value) - if cursor.value == "fixture-depth-3": - depth_3_call_count += 1 - return pages_by_cursor[cursor.value] - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - self.assertEqual(depth_3_call_count, 1) - self.assertEqual(result.fetched_cursor_count, 7) - # No loop: total callback invocations equal distinct cursors - # (7). If the echo were not skipped, the count would exceed - # 7 or the queue would not drain. - self.assertEqual(len(callback_calls), 7) - self.assertEqual( - callback_calls.count("fixture-depth-3"), - 1, - ) - - def test_depth_seven_is_a_leaf_queue_exhausts_naturally(self) -> None: - # Depth-7's fixture exposes no eligible continuation cursor. - # The walker must terminate by the queue emptying naturally, - # not by any error or cap. - initial_page = self._initial_page_with_root_and_one_cursor() - pages_by_cursor = self._parse_all_depth_pages() - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - return pages_by_cursor[cursor.value] - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - ids = {p.post_id for p in result.posts} - self.assertIn(self.DEPTH_POST_IDS[6], ids) - # Depth-7 has no cursors, so the FIFO empties after the - # seventh fetch. fetched_cursor_count is exactly seven. - self.assertEqual(result.fetched_cursor_count, 7) - self.assertNotIn( - self.DEPTH_POST_IDS[6], - {c.value for c in pages_by_cursor["fixture-depth-7"].cursors}, - ) - - def test_no_cursor_fast_path_initial_posts_unchanged(self) -> None: - # An initial page with zero cursors is the fast path: - # the walker makes no callback calls, leaves initial posts - # unchanged, and reports fetched_cursor_count == 0. - root_tweet = _tweet_result_with_user( - rest_id=ROOT_ID, - author_legacy="fixture-root-author", - handle_legacy="fixture-root-handle", - user_core={ - "name": "fixture-root-author", - "screen_name": "fixture-root-handle", - }, - full_text="fixture-root-text", - in_reply_to=None, - ) - initial_wire = _page([ - { - "type": "TimelineAddEntries", - "entries": [ - { - "entryId": "initial-root-only", - "content": { - "entryType": "TimelineTimelineItem", - "itemContent": _wrap_tweet_result(root_tweet), - }, - }, - ], - }, - ]) - initial_page = parse_tweet_detail_page( - initial_wire, focal_id=ROOT_ID, - ) - self.assertEqual(initial_page.cursors, ()) - - callback_calls: list[str] = [] - - def fetch_page(cursor: Cursor) -> TweetDetailPage: - callback_calls.append(cursor.value) - raise AssertionError( - f"callback must not fire; got cursor {cursor.value!r}", - ) - - result = walk_continuation_queue( - initial_page, fetch_page=fetch_page, - ) - - self.assertEqual(callback_calls, []) - self.assertEqual(result.fetched_cursor_count, 0) - ids = {p.post_id for p in result.posts} - self.assertEqual(ids, {ROOT_ID}) - self.assertEqual(len(result.posts), 1) - - -class RequestGovernorFirstCallTests(unittest.TestCase): - """Task 8 (item 1): the very first call to ``governor.issue`` - must not call ``sleeper.sleep`` (the operator has not yet - issued a prior request; there is nothing to pace against) and - must return the operation's ``(status, body)`` tuple verbatim. - """ - - def test_first_call_does_not_sleep_and_returns_result_verbatim(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - # Sentinel: would explode if invoked. - def operation() -> tuple[int, str]: - return (200, "first-call-body") - - result = governor.issue("bootstrap_homepage", operation) - - # Result is passed through unchanged. - self.assertEqual(result, (200, "first-call-body")) - # No sleep fired on the first call. - self.assertEqual(sleeper.durations, []) - # The first attempt is counted and recorded exactly once. - self.assertEqual(governor.count, 1) - self.assertEqual(governor.roles, ("bootstrap_homepage",)) - - -class RequestGovernorFourRoleSequenceTests(unittest.TestCase): - """Task 8 (item 2): the human-paced sequence for the locked - four-provider call shape is four roles with three sleeps in - between (the first call does not sleep). The exact role - tuple is the ordered log of issued calls. - """ - - SEQUENCE: tuple[tuple[str, tuple[int, str]], ...] = ( - ("bootstrap_homepage", (200, "home")), - ("bootstrap_js", (200, "js")), - ("tip_tweet_detail", (200, "tip")), - ("root_tweet_detail", (200, "root")), - ) - - def test_four_roles_three_sleeps_with_exact_role_order(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - - for role, body in self.SEQUENCE: - def operation( - _body: tuple[int, str] = body, - ) -> tuple[int, str]: - return _body - governor.issue(role, operation) - - # Three sleeps between the four calls; each value is - # exactly ``min_delay`` (5.0). - self.assertEqual(sleeper.durations, [5.0, 5.0, 5.0]) - # Roles preserve insertion order. - self.assertEqual( - governor.roles, - ("bootstrap_homepage", "bootstrap_js", - "tip_tweet_detail", "root_tweet_detail"), - ) - self.assertEqual(governor.count, 4) - - -class RequestGovernorMaxRequestsTests(unittest.TestCase): - """Task 8 (item 3): with ``max_requests=8`` the governor must - execute exactly 8 operations, sleep 7 times, and on the ninth - attempt raise ``RequestBudgetExceeded`` without calling the - operation, without sleeping, and without appending a role. - """ - - def test_ninth_attempt_raises_without_calling_sentinel(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - # Eight innocuous operations that each return (200, ""). - for _ in range(8): - governor.issue("ok", lambda: (200, "")) - - sentinel_calls: list[int] = [] - - def sentinel() -> tuple[int, str]: - sentinel_calls.append(1) - return (200, "should-never-run") - - with self.assertRaises(RequestBudgetExceeded): - governor.issue("ninth", sentinel) - - # Exactly seven sleeps (one between each of the eight - # successful calls; none on the first, none on the ninth). - self.assertEqual(len(sleeper.durations), 7) - self.assertEqual(sleeper.durations, [5.0] * 7) - # The sentinel operation is never invoked. - self.assertEqual(sentinel_calls, []) - # No extra role was appended for the failed ninth attempt. - self.assertEqual(len(governor.roles), 8) - # ``RequestBudgetExceeded`` is a ``RuntimeError``. - self.assertTrue( - issubclass(RequestBudgetExceeded, RuntimeError), - msg="RequestBudgetExceeded must subclass RuntimeError", - ) - - -class RequestGovernorOperationExceptionTests(unittest.TestCase): - """Task 8 (item 4): if the wrapped operation raises, the - governor must propagate the exception immediately, must not - retry, must not catch, must not strip the role from the count - (the attempted call still counts), and must not append a - second sleep entry. - """ - - def test_operation_exception_propagates_role_counted_no_retry(self) -> None: - class CallbackError(RuntimeError): - pass - - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - call_count: list[int] = [] - - def operation() -> tuple[int, str]: - call_count.append(1) - raise CallbackError("simulated provider failure") - - with self.assertRaises(CallbackError): - governor.issue("bootstrap_homepage", operation) - - # Operation called exactly once; no retry from the governor. - self.assertEqual(call_count, [1]) - # The attempted call is counted (it would have been a - # budgeted request). - self.assertEqual(governor.count, 1) - self.assertEqual(governor.roles, ("bootstrap_homepage",)) - # No sleep was issued: this is the first call. - self.assertEqual(sleeper.durations, []) - - def test_operation_exception_after_pacing_still_propagates_no_extra_sleep( - self, - ) -> None: - # Two-issue sequence: the first call succeeds, the second - # raises. The governor must not append a second sleep after - # the failing call, and the failing call must remain - # counted. - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - - class CallbackError(RuntimeError): - pass - - governor.issue("bootstrap_homepage", lambda: (200, "ok")) - with self.assertRaises(CallbackError): - governor.issue( - "bootstrap_js", - lambda: (_ for _ in ()).throw(CallbackError("boom")), - ) - # Exactly one sleep: the pace between the two attempted - # calls. No sleep is appended on the failing path. - self.assertEqual(sleeper.durations, [5.0]) - # Both attempted calls are counted. - self.assertEqual(governor.count, 2) - self.assertEqual( - governor.roles, ("bootstrap_homepage", "bootstrap_js"), - ) - - -class RequestGovernorValidationTests(unittest.TestCase): - """Task 8 (item 5): construction rejects ``max_requests < 1`` - and ``min_delay < 0``; ``issue`` rejects empty roles. The - governor fails closed at construction and at every call. - """ - - def test_constructor_rejects_zero_max_requests(self) -> None: - sleeper = RecordingSleeper() - with self.assertRaises(ValueError): - RequestGovernor(max_requests=0, min_delay=5.0, sleeper=sleeper) - - def test_constructor_rejects_negative_max_requests(self) -> None: - sleeper = RecordingSleeper() - with self.assertRaises(ValueError): - RequestGovernor(max_requests=-1, min_delay=5.0, sleeper=sleeper) - - def test_constructor_rejects_negative_min_delay(self) -> None: - sleeper = RecordingSleeper() - with self.assertRaises(ValueError): - RequestGovernor(max_requests=8, min_delay=-0.1, sleeper=sleeper) - - def test_constructor_accepts_zero_min_delay(self) -> None: - # min_delay == 0 is allowed; the operator may opt out of - # pacing entirely. The first call still does not sleep and - # later calls sleep(0.0). - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=0.0, sleeper=sleeper, - ) - governor.issue("a", lambda: (200, "")) - governor.issue("b", lambda: (200, "")) - self.assertEqual(sleeper.durations, [0.0]) - - def test_issue_rejects_empty_role(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - with self.assertRaises(ValueError): - governor.issue("", lambda: (200, "")) - - def test_constructor_rejects_bool_max_requests(self) -> None: - # ``True`` is an ``int`` subclass; the governor must still - # reject it. A bool is not a valid request budget. - sleeper = RecordingSleeper() - with self.assertRaises(ValueError): - RequestGovernor(max_requests=True, min_delay=5.0, sleeper=sleeper) - - def test_constructor_rejects_bool_min_delay(self) -> None: - # Same reasoning: ``True`` is an ``int`` but not a valid - # delay. The governor fails closed. - sleeper = RecordingSleeper() - with self.assertRaises(ValueError): - RequestGovernor(max_requests=8, min_delay=True, sleeper=sleeper) - - def test_constructor_rejects_none_sleeper(self) -> None: - # The sleeper is required: there is no default. The - # governor must reject ``None`` at construction rather - # than blow up on the first call. - with self.assertRaises(ValueError): - RequestGovernor(max_requests=8, min_delay=5.0, sleeper=None) - - def test_issue_rejects_non_string_role(self) -> None: - # Roles must be strings. An int, ``None``, or any other - # type is rejected before any state mutation. The first - # call to ``issue`` is the validation target. - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - with self.assertRaises(ValueError): - governor.issue(123, lambda: (200, "")) # type: ignore[arg-type] - # No role was appended for the rejected call. - self.assertEqual(governor.count, 0) - self.assertEqual(governor.roles, ()) - # And ``None`` is rejected too. - with self.assertRaises(ValueError): - governor.issue(None, lambda: (200, "")) # type: ignore[arg-type] - self.assertEqual(governor.count, 0) - self.assertEqual(governor.roles, ()) - - -class RequestGovernorSleeperExceptionTests(unittest.TestCase): - """Task 8 (item 3 follow-up): if ``sleeper.sleep`` raises - between two attempted calls, the governor must propagate - the exception, must NOT append the second role, must NOT - invoke the second operation, and must leave the count at - the post-first-call value. The governor does not catch, - retry, or strip the role. - """ - - def test_sleeper_exception_prevents_second_role_append_and_operation( - self, - ) -> None: - class SleeperFailure(RuntimeError): - pass - - class RaisingSleeper: - """A :class:`Sleeper`-shaped fake that raises on the - FIRST ``sleep`` call only. The first attempted call - does not sleep (the governor skips the sleep when - ``count == 0``), so the first call succeeds; the - second attempted call invokes ``sleep`` for the first - time and that invocation raises.""" - - def __init__(self) -> None: - self.sleep_calls: list[float] = [] - - def sleep(self, seconds: float) -> None: - self.sleep_calls.append(seconds) - raise SleeperFailure("simulated sleeper failure") - - sleeper = RaisingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - - # First call succeeds: no sleep on the first call, the - # operation runs, the role is appended. - first_result = governor.issue( - "bootstrap_homepage", lambda: (200, "home"), - ) - self.assertEqual(first_result, (200, "home")) - self.assertEqual(governor.count, 1) - self.assertEqual(governor.roles, ("bootstrap_homepage",)) - - # The second attempted call triggers a sleep; the sleeper - # raises before the role is appended and before the - # operation is invoked. - second_op_calls: list[int] = [] - - def second_operation() -> tuple[int, str]: - second_op_calls.append(1) - return (200, "should-never-run") - - with self.assertRaises(SleeperFailure): - governor.issue("bootstrap_js", second_operation) - - # The second role is not appended. - self.assertEqual(governor.roles, ("bootstrap_homepage",)) - # The second operation is not invoked. - self.assertEqual(second_op_calls, []) - # The count remains at the post-first-call value. - self.assertEqual(governor.count, 1) - # The sleeper was called exactly once with ``min_delay`` - # before raising on the first (and only) invocation — - # the governor does not retry the sleep. - self.assertEqual(sleeper.sleep_calls, [5.0]) - - -class RequestGovernorIndependentInstancesTests(unittest.TestCase): - """Task 8 (item 6): two governor instances do not share - state. Counts, roles, and sleep histories are independent. - """ - - def test_independent_governors_do_not_share_state(self) -> None: - sleeper_a = RecordingSleeper() - sleeper_b = RecordingSleeper() - a = RequestGovernor(max_requests=8, min_delay=5.0, sleeper=sleeper_a) - b = RequestGovernor(max_requests=8, min_delay=5.0, sleeper=sleeper_b) - - a.issue("bootstrap_homepage", lambda: (200, "")) - a.issue("bootstrap_js", lambda: (200, "")) - - b.issue("tip_tweet_detail", lambda: (200, "")) - - self.assertEqual(a.count, 2) - self.assertEqual( - a.roles, ("bootstrap_homepage", "bootstrap_js"), - ) - self.assertEqual(b.count, 1) - self.assertEqual(b.roles, ("tip_tweet_detail",)) - - # Sleepers tracked their respective governors; neither - # saw the other's calls. - self.assertEqual(sleeper_a.durations, [5.0]) - self.assertEqual(sleeper_b.durations, []) - - def test_sleeper_protocol_is_runtime_checkable_shape(self) -> None: - # The :class:`Sleeper` protocol declares ``sleep``; the - # Task 1 ``RecordingSleeper`` satisfies it. This pins the - # duck-typed surface the governor depends on. - sleeper = RecordingSleeper() - self.assertTrue(hasattr(sleeper, "sleep")) - self.assertTrue(callable(sleeper.sleep)) - - -class IssueProviderRequestSuccessTests(unittest.TestCase): - """Task 9: 2xx responses return the body string verbatim and - invoke the operation exactly once through the governor.""" - - def test_200_returns_body_one_call_count_one(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - body = "ok-body-sentinel" - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (200, body) - - result = issue_provider_request( - governor, role="bootstrap_homepage", operation=op, - ) - self.assertEqual(result, body) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - self.assertEqual(governor.roles, ("bootstrap_homepage",)) - - def test_204_returns_body_one_call_count_one(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - body = "no-content-sentinel" - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (204, body) - - result = issue_provider_request( - governor, role="bootstrap_js", operation=op, - ) - self.assertEqual(result, body) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - -class IssueProviderRequestAuthErrorTests(unittest.TestCase): - """Task 9: 401 and 403 responses raise :class:`AuthError`, - the operation is called exactly once, and no retry occurs.""" - - def test_401_raises_auth_error_no_retry(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (401, "") - - with self.assertRaises(AuthError): - issue_provider_request( - governor, role="tip_tweet_detail", operation=op, - ) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - def test_403_raises_auth_error_no_retry(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (403, "") - - with self.assertRaises(AuthError): - issue_provider_request( - governor, role="root_tweet_detail", operation=op, - ) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - -class IssueProviderRequestRateLimitTests(unittest.TestCase): - """Task 9: 429 responses raise :class:`RateLimitError`, - the operation is called exactly once, and no retry occurs.""" - - def test_429_raises_rate_limit_error_no_retry(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (429, "") - - with self.assertRaises(RateLimitError): - issue_provider_request( - governor, role="continuation_tweet_detail", operation=op, - ) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - -class IssueProviderRequestGenericErrorTests(unittest.TestCase): - """Task 9: any non-2xx that is not 401/403/429 raises - :class:`ProviderError`, the operation is called exactly once, - and no retry occurs. Statuses covered: 400, 404, 500, 503.""" - - def _assert_provider_error( - self, status: int, role: str, - ) -> tuple[RequestGovernor, list[int]]: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - return (status, "") - - with self.assertRaises(ProviderError): - issue_provider_request( - governor, role=role, operation=op, - ) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - return governor, call_count - - def test_400_raises_provider_error_no_retry(self) -> None: - self._assert_provider_error(400, "tip_tweet_detail") - - def test_404_raises_provider_error_no_retry(self) -> None: - self._assert_provider_error(404, "root_tweet_detail") - - def test_500_raises_provider_error_no_retry(self) -> None: - self._assert_provider_error(500, "bootstrap_homepage") - - def test_503_raises_provider_error_no_retry(self) -> None: - self._assert_provider_error(503, "continuation_tweet_detail") - - -class IssueProviderRequestTransportErrorTests(unittest.TestCase): - """Task 9: :class:`TimeoutError`, :class:`ConnectionError`, and - :class:`OSError` raised by the operation are translated into - :class:`ProviderError` chained from the original, with the - operation called exactly once.""" - - def test_timeout_error_becomes_provider_error_chained(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - sentinel = TimeoutError("simulated timeout sentinel") - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - raise sentinel - - with self.assertRaises(ProviderError) as ctx: - issue_provider_request( - governor, role="tip_tweet_detail", operation=op, - ) - self.assertIs(ctx.exception.__cause__, sentinel) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - def test_connection_error_becomes_provider_error_chained(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - sentinel = ConnectionError("simulated connection sentinel") - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - raise sentinel - - with self.assertRaises(ProviderError) as ctx: - issue_provider_request( - governor, role="tip_tweet_detail", operation=op, - ) - self.assertIs(ctx.exception.__cause__, sentinel) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - def test_oserror_becomes_provider_error_chained(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - sentinel = OSError("simulated os sentinel") - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - raise sentinel - - with self.assertRaises(ProviderError) as ctx: - issue_provider_request( - governor, role="tip_tweet_detail", operation=op, - ) - self.assertIs(ctx.exception.__cause__, sentinel) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - -class IssueProviderRequestNonProviderExceptionTests(unittest.TestCase): - """Task 9: arbitrary programmer/parser exceptions such as - :class:`ValueError` propagate unchanged and are NOT mislabelled - as a provider failure.""" - - def test_value_error_propagates_unchanged(self) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - sentinel = ValueError("simulated parser sentinel") - call_count: list[int] = [] - - def op() -> tuple[int, str]: - call_count.append(1) - raise sentinel - - with self.assertRaises(ValueError) as ctx: - issue_provider_request( - governor, role="tip_tweet_detail", operation=op, - ) - # Same instance propagates, not wrapped in ProviderError. - self.assertIs(ctx.exception, sentinel) - self.assertEqual(call_count, [1]) - self.assertEqual(governor.count, 1) - - -class IssueProviderRequestSecretFreeErrorMessagesTests(unittest.TestCase): - """Task 9: exception messages must NOT echo the response body, - URL, header value, or cursor value carried by the operation. - Status and exception type are the only diagnostic surfaces.""" - - def test_provider_error_message_does_not_leak_payload_sentinels( - self, - ) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - body_sentinel = "SENTINEL_BODY_abcdef0123456789" - url_sentinel = "https://SENTINEL_URL.example/x" - header_sentinel = "SENTINEL_HEADER_abcdef0123456789" - cursor_sentinel = "SENTINEL_CURSOR_abcdef0123456789" - - def op() -> tuple[int, str]: - return (503, body_sentinel) - - with self.assertRaises(ProviderError) as ctx: - issue_provider_request( - governor, - role=( - f"{url_sentinel} {header_sentinel} {cursor_sentinel}" - ), - operation=op, - ) - message = str(ctx.exception) - for needle in ( - body_sentinel, url_sentinel, header_sentinel, cursor_sentinel, - ): - self.assertNotIn(needle, message) - - def test_auth_error_message_does_not_leak_payload_sentinels( - self, - ) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - body_sentinel = "SENTINEL_BODY_abcdef0123456789" - url_sentinel = "https://SENTINEL_URL.example/x" - header_sentinel = "SENTINEL_HEADER_abcdef0123456789" - cursor_sentinel = "SENTINEL_CURSOR_abcdef0123456789" - - def op() -> tuple[int, str]: - return (401, body_sentinel) - - with self.assertRaises(AuthError) as ctx: - issue_provider_request( - governor, - role=( - f"{url_sentinel} {header_sentinel} {cursor_sentinel}" - ), - operation=op, - ) - message = str(ctx.exception) - for needle in ( - body_sentinel, url_sentinel, header_sentinel, cursor_sentinel, - ): - self.assertNotIn(needle, message) - - def test_rate_limit_error_message_does_not_leak_payload_sentinels( - self, - ) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - body_sentinel = "SENTINEL_BODY_abcdef0123456789" - url_sentinel = "https://SENTINEL_URL.example/x" - header_sentinel = "SENTINEL_HEADER_abcdef0123456789" - cursor_sentinel = "SENTINEL_CURSOR_abcdef0123456789" - - def op() -> tuple[int, str]: - return (429, body_sentinel) - - with self.assertRaises(RateLimitError) as ctx: - issue_provider_request( - governor, - role=( - f"{url_sentinel} {header_sentinel} {cursor_sentinel}" - ), - operation=op, - ) - message = str(ctx.exception) - for needle in ( - body_sentinel, url_sentinel, header_sentinel, cursor_sentinel, - ): - self.assertNotIn(needle, message) - - -class IssueProviderRequestBudgetExceededTests(unittest.TestCase): - """Task 9: :class:`RequestBudgetExceeded` must remain a - :class:`RuntimeError` and a :class:`FetchError`; the ninth - operation never runs when the budget is 8.""" - - def test_request_budget_exceeded_is_runtime_error_and_fetch_error( - self, - ) -> None: - self.assertTrue( - issubclass(RequestBudgetExceeded, RuntimeError), - msg="RequestBudgetExceeded must remain a RuntimeError", - ) - self.assertTrue( - issubclass(RequestBudgetExceeded, FetchError), - msg=( - "RequestBudgetExceeded must be a FetchError subclass " - "while remaining a RuntimeError" - ), - ) - - def test_ninth_operation_never_called_after_eight_successful( - self, - ) -> None: - sleeper = RecordingSleeper() - governor = RequestGovernor( - max_requests=8, min_delay=5.0, sleeper=sleeper, - ) - for i in range(8): - governor.issue( - f"ok-{i}", lambda i=i: (200, f"body-{i}"), - ) - - sentinel_calls: list[int] = [] - - def sentinel() -> tuple[int, str]: - sentinel_calls.append(1) - return (200, "should-never-run") - - with self.assertRaises(RequestBudgetExceeded): - issue_provider_request( - governor, role="ninth", operation=sentinel, - ) - self.assertEqual(sentinel_calls, []) - self.assertEqual(governor.count, 8) - self.assertEqual( - governor.roles, - tuple(f"ok-{i}" for i in range(8)), - ) - - -if __name__ == "__main__": - unittest.main() diff --git a/secrets/README.md b/secrets/README.md index a419035..640022d 100644 --- a/secrets/README.md +++ b/secrets/README.md @@ -1,21 +1,5 @@ Local credentials. Gitignored except this file. -| File | Contents | -| --- | --- | -| `twitter_cookies.txt` | Netscape cookies for gallery-dl | -| `credentials.toml` | `[catbox] userhash`, `[backup] id` and `root` | - -Shape of `credentials.toml`: - -```toml -[catbox] -userhash = "" - -[backup] -id = "cozy" -root = "" -``` - The scripts read these files. They never print the values. Do not commit this folder. Do not paste the contents into chat or issues.