From 526816c10f2397bdc092bc1e2e45694a0232f009 Mon Sep 17 00:00:00 2001 From: Ed_ Date: Mon, 24 Aug 2026 21:54:56 -0400 Subject: [PATCH] rediculous... --- scripts/twitter/fetch_thread.py | 971 ++++++++ scripts/twitter/tests/fetch_thread_helpers.py | 206 ++ .../tests/fixtures/fetch_thread/.gitkeep | 0 .../fetch_thread/continuation_depth_1.json | 58 + .../fetch_thread/continuation_depth_2.json | 58 + .../fetch_thread/continuation_depth_3.json | 66 + .../fetch_thread/continuation_depth_4.json | 58 + .../fetch_thread/continuation_depth_5.json | 58 + .../fetch_thread/continuation_depth_6.json | 58 + .../fetch_thread/continuation_depth_7.json | 50 + .../fetch_thread/continuation_tip_branch.json | 89 + .../fetch_thread/root_tweet_detail.json | 539 +++++ .../fetch_thread/tip_tweet_detail.json | 277 +++ scripts/twitter/tests/test_fetch_thread.py | 2025 +++++++++++++++++ 14 files changed, 4513 insertions(+) create mode 100644 scripts/twitter/fetch_thread.py create mode 100644 scripts/twitter/tests/fetch_thread_helpers.py create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/.gitkeep create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_1.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_2.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_3.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_4.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_5.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_6.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_7.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/continuation_tip_branch.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/root_tweet_detail.json create mode 100644 scripts/twitter/tests/fixtures/fetch_thread/tip_tweet_detail.json create mode 100644 scripts/twitter/tests/test_fetch_thread.py diff --git a/scripts/twitter/fetch_thread.py b/scripts/twitter/fetch_thread.py new file mode 100644 index 0000000..606d677 --- /dev/null +++ b/scripts/twitter/fetch_thread.py @@ -0,0 +1,971 @@ +"""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/tests/fetch_thread_helpers.py b/scripts/twitter/tests/fetch_thread_helpers.py new file mode 100644 index 0000000..d539ae9 --- /dev/null +++ b/scripts/twitter/tests/fetch_thread_helpers.py @@ -0,0 +1,206 @@ +"""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/tests/fixtures/fetch_thread/.gitkeep b/scripts/twitter/tests/fixtures/fetch_thread/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_1.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_1.json new file mode 100644 index 0000000..3872092 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_1.json @@ -0,0 +1,58 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_2.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_2.json new file mode 100644 index 0000000..3715a8d --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_2.json @@ -0,0 +1,58 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_3.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_3.json new file mode 100644 index 0000000..668fd40 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_3.json @@ -0,0 +1,66 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_4.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_4.json new file mode 100644 index 0000000..4171f62 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_4.json @@ -0,0 +1,58 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_5.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_5.json new file mode 100644 index 0000000..250a9e6 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_5.json @@ -0,0 +1,58 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_6.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_6.json new file mode 100644 index 0000000..255f2e4 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_6.json @@ -0,0 +1,58 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_depth_7.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_7.json new file mode 100644 index 0000000..3e997ae --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_depth_7.json @@ -0,0 +1,50 @@ +{ + "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/tests/fixtures/fetch_thread/continuation_tip_branch.json b/scripts/twitter/tests/fixtures/fetch_thread/continuation_tip_branch.json new file mode 100644 index 0000000..e009e52 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/continuation_tip_branch.json @@ -0,0 +1,89 @@ +{ + "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/tests/fixtures/fetch_thread/root_tweet_detail.json b/scripts/twitter/tests/fixtures/fetch_thread/root_tweet_detail.json new file mode 100644 index 0000000..b8c74a1 --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/root_tweet_detail.json @@ -0,0 +1,539 @@ +{ + "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 + } + } + } + } + } + } + ] + } + ] + } + } +} \ No newline at end of file diff --git a/scripts/twitter/tests/fixtures/fetch_thread/tip_tweet_detail.json b/scripts/twitter/tests/fixtures/fetch_thread/tip_tweet_detail.json new file mode 100644 index 0000000..1afde0f --- /dev/null +++ b/scripts/twitter/tests/fixtures/fetch_thread/tip_tweet_detail.json @@ -0,0 +1,277 @@ +{ + "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/tests/test_fetch_thread.py b/scripts/twitter/tests/test_fetch_thread.py new file mode 100644 index 0000000..82285fe --- /dev/null +++ b/scripts/twitter/tests/test_fetch_thread.py @@ -0,0 +1,2025 @@ +"""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()