| 1 | """xapi: the direct X API v2 backend and the shared v2 parser (U4). |
| 2 | |
| 3 | Every test mocks ``lib.http.get`` (or urlopen for the fixture seam); no test |
| 4 | makes a live X call. The dummy token below is obvious fake data and must |
| 5 | never appear in any error string, source_log line, or recorded fixture. |
| 6 | """ |
| 7 | |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import io |
| 11 | import json |
| 12 | import threading |
| 13 | from contextlib import redirect_stderr |
| 14 | from datetime import datetime, timezone |
| 15 | from unittest import mock |
| 16 | from unittest.mock import MagicMock |
| 17 | |
| 18 | import pytest |
| 19 | |
| 20 | from lib import env, health, http, pipeline, schema, x_api, xurl_x |
| 21 | |
| 22 | DUMMY_TOKEN = "dummy-x-bearer-secret-000" |
| 23 | ACCOUNT_ID = "acct-1234567890" |
| 24 | FIXED_NOW = datetime(2026, 9, 8, 12, 0, 0, tzinfo=timezone.utc) |
| 25 | FROM = "2026-08-10" |
| 26 | TO = "2026-09-08" |
| 27 | |
| 28 | |
| 29 | def _tweet(id_, text, author_id="u1", created_at="2026-08-20T12:00:00Z", metrics=None, **extra): |
| 30 | t = {"id": id_, "text": text, "author_id": author_id} |
| 31 | if created_at: |
| 32 | t["created_at"] = created_at |
| 33 | if metrics is not None: |
| 34 | t["public_metrics"] = metrics |
| 35 | t.update(extra) |
| 36 | return t |
| 37 | |
| 38 | |
| 39 | def _v2(tweets, users=None, next_token=None): |
| 40 | resp = {"data": tweets, "meta": {"result_count": len(tweets)}} |
| 41 | if users: |
| 42 | resp["includes"] = {"users": users} |
| 43 | if next_token: |
| 44 | resp["meta"]["next_token"] = next_token |
| 45 | return resp |
| 46 | |
| 47 | |
| 48 | def _users(*pairs): |
| 49 | return [{"id": uid, "username": name} for uid, name in pairs] |
| 50 | |
| 51 | |
| 52 | def _http_error(status, body="", message=None): |
| 53 | return http.HTTPError(message or f"HTTP {status}: Error", status_code=status, body=body) |
| 54 | |
| 55 | |
| 56 | @pytest.fixture |
| 57 | def fixed_now(monkeypatch): |
| 58 | monkeypatch.setattr(x_api, "_utcnow", lambda: FIXED_NOW) |
| 59 | return FIXED_NOW |
| 60 | |
| 61 | |
| 62 | @pytest.fixture |
| 63 | def get_mock(monkeypatch): |
| 64 | m = MagicMock(name="http.get") |
| 65 | monkeypatch.setattr(x_api.http, "get", m) |
| 66 | return m |
| 67 | |
| 68 | |
| 69 | def _params(call): |
| 70 | return call.kwargs.get("params") or {} |
| 71 | |
| 72 | |
| 73 | def _url(call): |
| 74 | return call.args[0] if call.args else call.kwargs["url"] |
| 75 | |
| 76 | |
| 77 | # --------------------------------------------------------------------------- |
| 78 | # parse_v2_response |
| 79 | # --------------------------------------------------------------------------- |
| 80 | |
| 81 | |
| 82 | class TestParseV2Response: |
| 83 | def test_fixture_with_users_parses_all_fields(self): |
| 84 | metrics = { |
| 85 | "like_count": 42, "retweet_count": 10, "reply_count": 5, |
| 86 | "quote_count": 2, "bookmark_count": 7, "impression_count": 9000, |
| 87 | } |
| 88 | resp = _v2( |
| 89 | [_tweet("1956158892141441450", "Claude Code agents are great", metrics=metrics)], |
| 90 | users=_users(("u1", "alice")), |
| 91 | ) |
| 92 | items = x_api.parse_v2_response(resp, "Claude Code", (FROM, TO)) |
| 93 | assert len(items) == 1 |
| 94 | item = items[0] |
| 95 | assert item["id"] == "XAPI1" |
| 96 | assert item["post_id"] == "1956158892141441450" |
| 97 | assert item["author_handle"] == "alice" |
| 98 | assert item["url"] == "https://x.com/alice/status/1956158892141441450" |
| 99 | assert item["date"] == "2026-08-20" |
| 100 | assert item["engagement"] == { |
| 101 | "likes": 42, "reposts": 10, "replies": 5, "quotes": 2, |
| 102 | "bookmarks": 7, "views": 9000, |
| 103 | } |
| 104 | assert item["why_relevant"] == "" |
| 105 | assert item["relevance"] > 0.5 |
| 106 | assert item["mentioned_handles"] == [] |
| 107 | |
| 108 | def test_without_users_keeps_item_with_i_status_url(self): |
| 109 | resp = _v2([_tweet("999", "text here", author_id="unknown")]) |
| 110 | items = x_api.parse_v2_response(resp, "text", (FROM, TO)) |
| 111 | assert len(items) == 1 |
| 112 | assert items[0]["url"] == "https://x.com/i/status/999" |
| 113 | assert items[0]["author_handle"] == "" |
| 114 | |
| 115 | def test_note_tweet_replaces_truncated_text(self): |
| 116 | resp = _v2([_tweet("5", "truncated...", note_tweet={"text": "the full long-form post text"})]) |
| 117 | items = x_api.parse_v2_response(resp, "", None) |
| 118 | assert items[0]["text"] == "the full long-form post text" |
| 119 | |
| 120 | def test_url_is_built_from_id_never_from_response_url_fields(self): |
| 121 | resp = _v2( |
| 122 | [_tweet("7", "hi", url="https://evil.example/phish", |
| 123 | entities={"urls": [{"expanded_url": "https://evil.example/x"}]})], |
| 124 | users=_users(("u1", "bob")), |
| 125 | ) |
| 126 | items = x_api.parse_v2_response(resp, "", None) |
| 127 | assert items[0]["url"] == "https://x.com/bob/status/7" |
| 128 | assert "evil.example" not in json.dumps(items) |
| 129 | |
| 130 | def test_username_outside_handle_grammar_falls_back_to_i_status(self): |
| 131 | resp = _v2([_tweet("8", "hi")], users=[{"id": "u1", "username": "bad name/../x"}]) |
| 132 | items = x_api.parse_v2_response(resp, "", None) |
| 133 | assert items[0]["author_handle"] == "" |
| 134 | assert items[0]["url"] == "https://x.com/i/status/8" |
| 135 | |
| 136 | def test_leading_mentions_become_mentioned_handles(self): |
| 137 | resp = _v2([_tweet("9", "@steipete @alice thanks for this")]) |
| 138 | items = x_api.parse_v2_response(resp, "", None) |
| 139 | assert items[0]["mentioned_handles"] == ["steipete", "alice"] |
| 140 | |
| 141 | def test_engagement_none_without_metrics_and_bookmarks_views_only_when_present(self): |
| 142 | resp = _v2([ |
| 143 | _tweet("1", "a"), |
| 144 | _tweet("2", "b", metrics={"like_count": 1, "retweet_count": 0, "reply_count": 0, "quote_count": 0}), |
| 145 | ]) |
| 146 | items = x_api.parse_v2_response(resp, "", None) |
| 147 | assert items[0]["engagement"] is None |
| 148 | assert items[1]["engagement"] == {"likes": 1, "reposts": 0, "replies": 0, "quotes": 0} |
| 149 | |
| 150 | def test_non_numeric_ids_are_dropped(self): |
| 151 | resp = _v2([_tweet("abc", "x"), _tweet("12", "y")]) |
| 152 | items = x_api.parse_v2_response(resp, "", None) |
| 153 | assert [i["post_id"] for i in items] == ["12"] |
| 154 | |
| 155 | def test_window_drops_dated_items_outside_it(self): |
| 156 | resp = _v2([ |
| 157 | _tweet("1", "old", created_at="2026-07-01T00:00:00Z"), |
| 158 | _tweet("2", "in", created_at="2026-08-20T00:00:00Z"), |
| 159 | ]) |
| 160 | items = x_api.parse_v2_response(resp, "", (FROM, TO)) |
| 161 | assert [i["post_id"] for i in items] == ["2"] |
| 162 | |
| 163 | def test_error_response_yields_nothing(self): |
| 164 | assert x_api.parse_v2_response({"error": "x"}, "", None) == [] |
| 165 | assert x_api.parse_v2_response({}, "", None) == [] |
| 166 | |
| 167 | def test_xurl_delegates_to_shared_parser_and_depth_config(self): |
| 168 | resp = _v2([_tweet("3", "hello")], users=_users(("u1", "alice"))) |
| 169 | items = xurl_x.parse_x_response(resp, topic="hello") |
| 170 | assert items[0]["id"] == "XURL1" |
| 171 | assert items[0]["url"] == "https://x.com/alice/status/3" |
| 172 | assert items[0]["post_id"] == "3" |
| 173 | assert xurl_x.DEPTH_CONFIG is x_api.DEPTH_CONFIG |
| 174 | assert x_api.DEPTH_CONFIG == {"quick": 10, "default": 30, "deep": 60} |
| 175 | |
| 176 | |
| 177 | # --------------------------------------------------------------------------- |
| 178 | # Query compilation (R8a) |
| 179 | # --------------------------------------------------------------------------- |
| 180 | |
| 181 | |
| 182 | class TestBuildQuery: |
| 183 | def test_wraps_core_in_one_quote_pair_and_appends_no_retweets(self): |
| 184 | q = x_api.build_query('Claude Code "agents"') |
| 185 | assert q == '"Claude Code agents" -is:retweet' |
| 186 | |
| 187 | def test_operator_injection_is_stripped(self): |
| 188 | q = x_api.build_query('foo" OR from:attacker since:2015-01-01 "') |
| 189 | assert q == '"foo" -is:retweet' |
| 190 | assert "from:" not in q and "since:" not in q and " OR " not in q |
| 191 | assert q.count('"') == 2 |
| 192 | |
| 193 | def test_negation_and_grouping_characters_are_stripped(self): |
| 194 | q = x_api.build_query("(Peter Steinberger) -steipete “quoted” [x] {y}") |
| 195 | assert q == '"Peter Steinberger quoted x y" -is:retweet' |
| 196 | |
| 197 | def test_600_char_topic_compiles_under_512_with_balanced_quotes(self): |
| 198 | topic = " ".join(f"word{i}" for i in range(100)) |
| 199 | assert len(topic) >= 600 |
| 200 | q = x_api.build_query(topic) |
| 201 | assert len(q) <= x_api.MAX_QUERY_CHARS |
| 202 | assert q.count('"') == 2 |
| 203 | assert q.endswith('" -is:retweet') |
| 204 | assert not q.startswith('"word') or q.startswith('"word0 ') |
| 205 | # Cut at a token boundary: the last kept token is intact. |
| 206 | core = q[1:q.index('" -is:retweet')] |
| 207 | assert all(tok.startswith("word") and tok[4:].isdigit() for tok in core.split()) |
| 208 | |
| 209 | def test_single_giant_token_is_truncated_not_dropped(self): |
| 210 | q = x_api.build_query("a" * 700) |
| 211 | assert 0 < len(q) <= x_api.MAX_QUERY_CHARS |
| 212 | assert q.count('"') == 2 |
| 213 | |
| 214 | def test_only_operators_yields_empty_query(self): |
| 215 | assert x_api.build_query("from:attacker OR -is:reply lang:en") == "" |
| 216 | |
| 217 | |
| 218 | # --------------------------------------------------------------------------- |
| 219 | # Request shape (KTD4) |
| 220 | # --------------------------------------------------------------------------- |
| 221 | |
| 222 | |
| 223 | class TestRequestShape: |
| 224 | def test_full_archive_request_shape(self, get_mock, fixed_now): |
| 225 | get_mock.return_value = _v2([_tweet("1", "Claude Code rocks")], users=_users(("u1", "a"))) |
| 226 | result = x_api.search_x(DUMMY_TOKEN, "Claude Code", FROM, TO, depth="quick") |
| 227 | assert "error" not in result |
| 228 | assert [i["id"] for i in result["items"]] == ["XAPI1"] |
| 229 | call = get_mock.call_args |
| 230 | assert _url(call) == "https://api.x.com/2/tweets/search/all" |
| 231 | p = _params(call) |
| 232 | assert p["query"] == '"Claude Code" -is:retweet' |
| 233 | assert p["start_time"] == "2026-08-10T00:00:00Z" |
| 234 | # to_date is "today" relative to the fixed clock: end_time keeps a |
| 235 | # 30 second safety margin before now. |
| 236 | assert p["end_time"] == "2026-09-08T11:59:30Z" |
| 237 | assert p["sort_order"] == "recency" |
| 238 | assert p["max_results"] == 10 |
| 239 | assert p["expansions"] == "author_id" |
| 240 | assert p["tweet.fields"] == "created_at,public_metrics,note_tweet,entities" |
| 241 | assert p["user.fields"] == "username" |
| 242 | assert "lang:" not in p["query"] |
| 243 | assert call.kwargs["headers"]["Authorization"] == f"Bearer {DUMMY_TOKEN}" |
| 244 | assert call.kwargs["timeout"] == 30 |
| 245 | assert call.kwargs["retries"] == 2 |
| 246 | |
| 247 | def test_past_end_date_uses_end_of_day(self, get_mock, fixed_now): |
| 248 | get_mock.return_value = _v2([]) |
| 249 | x_api.search_x(DUMMY_TOKEN, "topic", "2026-08-01", "2026-08-31", depth="quick") |
| 250 | assert _params(get_mock.call_args)["end_time"] == "2026-08-31T23:59:59Z" |
| 251 | |
| 252 | def test_max_results_clamped_to_10_and_100(self, get_mock, fixed_now, monkeypatch): |
| 253 | get_mock.return_value = _v2([]) |
| 254 | x_api.search_handles(["steipete"], "t", FROM, TO, count_per=3, token=DUMMY_TOKEN) |
| 255 | assert _params(get_mock.call_args)["max_results"] == 10 |
| 256 | monkeypatch.setitem(x_api.DEPTH_CONFIG, "deep", 150) |
| 257 | x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="deep") |
| 258 | assert _params(get_mock.call_args)["max_results"] == 100 |
| 259 | |
| 260 | def test_pagination_stops_at_depth_count(self, get_mock, fixed_now): |
| 261 | page = lambda start, n, nxt: _v2( |
| 262 | [_tweet(str(1000 + start + i), f"post {i}") for i in range(n)], next_token=nxt, |
| 263 | ) |
| 264 | get_mock.side_effect = [page(0, 20, "p2"), page(20, 20, "p3"), page(40, 20, "p4")] |
| 265 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="default") |
| 266 | assert get_mock.call_count == 2, "30 posts reached after two pages; the third is never fetched" |
| 267 | assert len(result["items"]) == 30 |
| 268 | assert "next_token" not in _params(get_mock.call_args_list[0]) |
| 269 | assert _params(get_mock.call_args_list[1])["next_token"] == "p2" |
| 270 | assert len({i["id"] for i in result["items"]}) == 30 |
| 271 | |
| 272 | def test_pagination_stops_at_the_wall_clock_deadline(self, get_mock, fixed_now, monkeypatch): |
| 273 | page = lambda start, n, nxt: _v2( |
| 274 | [_tweet(str(1000 + start + i), f"post {i}") for i in range(n)], next_token=nxt, |
| 275 | ) |
| 276 | get_mock.side_effect = [page(0, 20, "p2"), page(20, 20, "p3"), page(40, 20, "p4")] |
| 277 | clock = iter([0.0, x_api.DEADLINE_SECONDS + 1.0, x_api.DEADLINE_SECONDS + 2.0, x_api.DEADLINE_SECONDS + 3.0]) |
| 278 | monkeypatch.setattr(x_api.time, "monotonic", lambda: next(clock)) |
| 279 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="deep") |
| 280 | assert get_mock.call_count == 1, "the first page always runs; the second is skipped past the deadline" |
| 281 | assert len(result["items"]) == 20 |
| 282 | assert "error" not in result |
| 283 | assert result["warning"] == x_api.DEADLINE_DETAIL |
| 284 | |
| 285 | def test_deadline_reaches_the_transport_and_keeps_pages_collected_before_it(self, get_mock, fixed_now, monkeypatch): |
| 286 | """The lane deadline is the transport's wall deadline (no full 30s |
| 287 | timeout plus retries past it), and a deadline hit mid-walk returns |
| 288 | the pages already collected.""" |
| 289 | page = lambda start, n, nxt: _v2( |
| 290 | [_tweet(str(1000 + start + i), f"post {i}") for i in range(n)], next_token=nxt, |
| 291 | ) |
| 292 | monkeypatch.setattr(x_api.time, "monotonic", lambda: 100.0) |
| 293 | get_mock.side_effect = [page(0, 20, "p2"), http.DeadlineExceeded()] |
| 294 | result = x_api.search_handles(["steipete"], "t", FROM, TO, count_per=60, token=DUMMY_TOKEN, deadline=130.0) |
| 295 | assert get_mock.call_count == 2 |
| 296 | for call in get_mock.call_args_list: |
| 297 | assert call.kwargs["deadline_monotonic"] == 130.0 |
| 298 | assert len(result) == 20 and "error" not in result[0] |
| 299 | get_mock.reset_mock() |
| 300 | # The same stop on the topic search is a warning receipt, never a |
| 301 | # healthy-looking complete result. |
| 302 | get_mock.side_effect = [page(0, 20, "p2"), http.DeadlineExceeded()] |
| 303 | topic = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="deep") |
| 304 | assert len(topic["items"]) == 20 |
| 305 | assert topic["warning"] == x_api.DEADLINE_DETAIL |
| 306 | get_mock.reset_mock() |
| 307 | get_mock.side_effect = [http.DeadlineExceeded()] |
| 308 | assert x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick")["error"] == x_api.ERR_TIMED_OUT |
| 309 | |
| 310 | def test_pagination_stops_without_next_token(self, get_mock, fixed_now): |
| 311 | get_mock.return_value = _v2([_tweet("1", "only one")]) |
| 312 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="deep") |
| 313 | assert get_mock.call_count == 1 |
| 314 | assert len(result["items"]) == 1 |
| 315 | |
| 316 | def test_every_request_targets_api_x_com_and_carries_no_token_outside_the_header(self, get_mock, fixed_now): |
| 317 | get_mock.side_effect = [ |
| 318 | _http_error(403, body='{"reason":"client-not-enrolled"}'), |
| 319 | _v2([_tweet("1", "a")], next_token="n"), |
| 320 | _v2([_tweet("2", "b")]), |
| 321 | ] |
| 322 | x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 323 | x_api.search_mentions(["steipete"], FROM, TO, topic="t", token=DUMMY_TOKEN) |
| 324 | assert get_mock.call_count >= 3 |
| 325 | for call in get_mock.call_args_list: |
| 326 | url = _url(call) |
| 327 | assert url.startswith("https://api.x.com/2/tweets/search/") |
| 328 | assert DUMMY_TOKEN not in url |
| 329 | assert DUMMY_TOKEN not in json.dumps(_params(call)) |
| 330 | headers = dict(call.kwargs["headers"]) |
| 331 | headers.pop("Authorization") |
| 332 | assert DUMMY_TOKEN not in json.dumps(headers) |
| 333 | |
| 334 | def test_operator_injection_still_carries_engine_window(self, get_mock, fixed_now): |
| 335 | get_mock.return_value = _v2([]) |
| 336 | x_api.search_x(DUMMY_TOKEN, 'foo" OR from:attacker since:2015-01-01 "', FROM, TO) |
| 337 | p = _params(get_mock.call_args) |
| 338 | assert p["query"] == '"foo" -is:retweet' |
| 339 | assert p["start_time"] == "2026-08-10T00:00:00Z" |
| 340 | assert p["end_time"] == "2026-09-08T11:59:30Z" |
| 341 | |
| 342 | def test_empty_query_after_sanitizing_makes_no_request(self, get_mock, fixed_now): |
| 343 | result = x_api.search_x(DUMMY_TOKEN, "from:attacker OR", FROM, TO) |
| 344 | assert result["items"] == [] |
| 345 | assert result["error"] == x_api.ERR_EMPTY_QUERY |
| 346 | get_mock.assert_not_called() |
| 347 | |
| 348 | def test_missing_token_makes_no_request(self, get_mock): |
| 349 | result = x_api.search_x("", "topic", FROM, TO) |
| 350 | assert result["items"] == [] and result["error"] |
| 351 | get_mock.assert_not_called() |
| 352 | |
| 353 | |
| 354 | # --------------------------------------------------------------------------- |
| 355 | # 403 enrollment fallback (KTD4) |
| 356 | # --------------------------------------------------------------------------- |
| 357 | |
| 358 | |
| 359 | class TestEnrollmentFallback: |
| 360 | def test_enrollment_403_retries_recent_once_with_clamped_window(self, get_mock, fixed_now): |
| 361 | get_mock.side_effect = [ |
| 362 | _http_error(403, body='{"reason":"client-not-enrolled","detail":"...","client_id":"' + ACCOUNT_ID + '"}'), |
| 363 | _v2([_tweet("1", "recent post", created_at="2026-09-05T00:00:00Z")], users=_users(("u1", "a"))), |
| 364 | ] |
| 365 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 366 | assert get_mock.call_count == 2 |
| 367 | assert _url(get_mock.call_args_list[0]) == "https://api.x.com/2/tweets/search/all" |
| 368 | assert _url(get_mock.call_args_list[1]) == "https://api.x.com/2/tweets/search/recent" |
| 369 | p2 = _params(get_mock.call_args_list[1]) |
| 370 | assert p2["start_time"] == "2026-09-01T12:00:30Z", "start clamped to now minus 7 days" |
| 371 | assert p2["end_time"] == "2026-09-08T11:59:30Z" |
| 372 | assert "error" not in result |
| 373 | assert result["warning"] == x_api.TRUNCATION_DETAIL == "window truncated to 7 days" |
| 374 | assert len(result["items"]) == 1 |
| 375 | |
| 376 | def test_enrollment_fallback_keeps_start_when_window_already_recent(self, get_mock, fixed_now): |
| 377 | get_mock.side_effect = [ |
| 378 | _http_error(403, body="not enrolled"), |
| 379 | _v2([]), |
| 380 | ] |
| 381 | x_api.search_x(DUMMY_TOKEN, "topic", "2026-09-06", TO, depth="quick") |
| 382 | assert _params(get_mock.call_args_list[1])["start_time"] == "2026-09-06T00:00:00Z" |
| 383 | |
| 384 | def test_plain_403_does_not_retry(self, get_mock, fixed_now): |
| 385 | get_mock.side_effect = [_http_error(403, body="Forbidden")] |
| 386 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 387 | assert get_mock.call_count == 1 |
| 388 | assert result["items"] == [] |
| 389 | assert result["error"] == x_api.ERR_FORBIDDEN |
| 390 | assert http.classify_failure(message=result["error"]) == health.AUTH_FAILED |
| 391 | |
| 392 | def test_enrollment_fallback_with_window_before_the_floor_sends_no_request(self, get_mock, fixed_now): |
| 393 | """A window that ends before now-7d cannot be served by recent search: |
| 394 | no second request with start_time after end_time.""" |
| 395 | get_mock.side_effect = [_http_error(403, body="client-not-enrolled")] |
| 396 | result = x_api.search_x(DUMMY_TOKEN, "topic", "2026-08-01", "2026-08-20", depth="quick") |
| 397 | assert get_mock.call_count == 1 |
| 398 | assert result == {"items": [], "warning": x_api.TRUNCATION_DETAIL} |
| 399 | |
| 400 | def test_second_403_on_recent_is_a_fixed_forbidden_error(self, get_mock, fixed_now): |
| 401 | get_mock.side_effect = [ |
| 402 | _http_error(403, body="client-not-enrolled"), |
| 403 | _http_error(403, body="client-not-enrolled " + DUMMY_TOKEN), |
| 404 | ] |
| 405 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 406 | assert get_mock.call_count == 2 |
| 407 | assert result["error"] == x_api.ERR_FORBIDDEN |
| 408 | |
| 409 | |
| 410 | # --------------------------------------------------------------------------- |
| 411 | # Fixed-string errors (R8, KTD6) |
| 412 | # --------------------------------------------------------------------------- |
| 413 | |
| 414 | # A body that echoes the token and an account id but carries no status |
| 415 | # marker of its own, so the status code alone decides the fixed string. |
| 416 | _LEAKY_BODY = json.dumps({ |
| 417 | "title": "Client Error", |
| 418 | "detail": f"request for account {ACCOUNT_ID} rejected; token {DUMMY_TOKEN}", |
| 419 | "client_id": ACCOUNT_ID, |
| 420 | }) |
| 421 | |
| 422 | |
| 423 | def _run_capturing_stderr(fn): |
| 424 | buf = io.StringIO() |
| 425 | with redirect_stderr(buf): |
| 426 | with mock.patch("sys.stderr.isatty", return_value=False, create=True): |
| 427 | out = fn() |
| 428 | return out, buf.getvalue() |
| 429 | |
| 430 | |
| 431 | class TestFixedErrors: |
| 432 | @pytest.mark.parametrize("status,expected,state", [ |
| 433 | (402, x_api.ERR_PAYMENT_REQUIRED, health.PAYMENT_REQUIRED), |
| 434 | (403, x_api.ERR_FORBIDDEN, health.AUTH_FAILED), |
| 435 | (401, x_api.ERR_UNAUTHORIZED, health.AUTH_FAILED), |
| 436 | (429, x_api.ERR_RATE_LIMITED, health.RATE_LIMITED), |
| 437 | ]) |
| 438 | def test_status_maps_to_fixed_string_without_body(self, get_mock, fixed_now, status, expected, state): |
| 439 | get_mock.side_effect = _http_error(status, body=_LEAKY_BODY, message=f"HTTP {status}: {ACCOUNT_ID}") |
| 440 | result, stderr = _run_capturing_stderr( |
| 441 | lambda: x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 442 | ) |
| 443 | assert result["items"] == [] |
| 444 | assert result["error"] == expected |
| 445 | assert http.classify_failure(message=result["error"]) == state |
| 446 | for leak in (DUMMY_TOKEN, ACCOUNT_ID, "Client Error", "request for account"): |
| 447 | assert leak not in result["error"] |
| 448 | assert leak not in stderr |
| 449 | assert "[xapi]" in stderr or "xapi" in stderr |
| 450 | |
| 451 | def test_fixed_strings_are_the_planned_literals(self): |
| 452 | assert x_api.ERR_PAYMENT_REQUIRED == "xapi: payment required (X API credits exhausted)" |
| 453 | assert x_api.ERR_UNAUTHORIZED == "xapi: unauthorized (bearer token rejected)" |
| 454 | assert x_api.ERR_FORBIDDEN == "xapi: forbidden (bearer token lacks access)" |
| 455 | assert x_api.ERR_RATE_LIMITED == "xapi: rate limit exceeded (X API)" |
| 456 | assert x_api.ERR_TIMED_OUT == "xapi: timed out" |
| 457 | |
| 458 | def test_credit_marker_in_body_maps_to_payment_required_on_any_status(self, get_mock, fixed_now): |
| 459 | get_mock.side_effect = _http_error(403, body='{"detail":"insufficient credits"}') |
| 460 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 461 | assert result["error"] == x_api.ERR_PAYMENT_REQUIRED |
| 462 | |
| 463 | def test_other_http_status_is_fixed_http_n(self, get_mock, fixed_now): |
| 464 | get_mock.side_effect = _http_error(503, body=_LEAKY_BODY) |
| 465 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 466 | assert result["error"] == "xapi: http 503" |
| 467 | |
| 468 | def test_timeout_is_fixed_string(self, get_mock, fixed_now): |
| 469 | get_mock.side_effect = http.HTTPError( |
| 470 | f"URL Error: timed out reading {ACCOUNT_ID}", outcome_state=health.TIMEOUT, |
| 471 | ) |
| 472 | result = x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 473 | assert result["error"] == x_api.ERR_TIMED_OUT |
| 474 | assert http.classify_failure(message=result["error"]) == health.TIMEOUT |
| 475 | |
| 476 | def test_unexpected_exception_never_carries_its_message(self, get_mock, fixed_now): |
| 477 | get_mock.side_effect = ValueError(f"boom {DUMMY_TOKEN} {ACCOUNT_ID}") |
| 478 | result, stderr = _run_capturing_stderr( |
| 479 | lambda: x_api.search_x(DUMMY_TOKEN, "topic", FROM, TO, depth="quick") |
| 480 | ) |
| 481 | assert result["error"] == "xapi: request failed (ValueError)" |
| 482 | assert DUMMY_TOKEN not in stderr and ACCOUNT_ID not in stderr |
| 483 | |
| 484 | @pytest.mark.parametrize("status,state", [(402, health.PAYMENT_REQUIRED), (403, health.AUTH_FAILED)]) |
| 485 | def test_pipeline_outcome_carries_only_the_fixed_string(self, get_mock, fixed_now, status, state): |
| 486 | get_mock.side_effect = _http_error(status, body=_LEAKY_BODY, message=f"HTTP {status}: {ACCOUNT_ID}") |
| 487 | sq = schema.SubQuery(label="primary", search_query="q", ranking_query="q?", sources=["x"]) |
| 488 | runtime = schema.ProviderRuntime( |
| 489 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", |
| 490 | x_search_backend=None, |
| 491 | ) |
| 492 | |
| 493 | def run(): |
| 494 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi"]): |
| 495 | with pytest.raises(pipeline.SourceRunError) as ctx: |
| 496 | pipeline._retrieve_stream( |
| 497 | topic="q", subquery=sq, source="x", |
| 498 | config={"X_BEARER_TOKEN": DUMMY_TOKEN}, |
| 499 | depth="quick", date_range=(FROM, TO), |
| 500 | runtime=runtime, mock=False, |
| 501 | ) |
| 502 | return ctx.value |
| 503 | |
| 504 | exc, stderr = _run_capturing_stderr(run) |
| 505 | assert exc.outcome_state == state |
| 506 | rendered = str(exc) |
| 507 | assert DUMMY_TOKEN not in rendered and ACCOUNT_ID not in rendered |
| 508 | assert DUMMY_TOKEN not in stderr and ACCOUNT_ID not in stderr |
| 509 | detail_state, attempted = pipeline._classify_source_failure(exc) |
| 510 | assert (detail_state, attempted) == (state, True) |
| 511 | |
| 512 | |
| 513 | # --------------------------------------------------------------------------- |
| 514 | # Handle lanes (R7) |
| 515 | # --------------------------------------------------------------------------- |
| 516 | |
| 517 | |
| 518 | class TestHandleLanes: |
| 519 | def test_from_lane_query_excludes_topic(self, get_mock, fixed_now): |
| 520 | get_mock.return_value = _v2( |
| 521 | [_tweet("1", "anything", created_at="2026-08-20T00:00:00Z")], users=_users(("u1", "steipete")), |
| 522 | ) |
| 523 | items = x_api.search_handles(["@steipete"], "Grok 4", FROM, TO, count_per=8, token=DUMMY_TOKEN) |
| 524 | p = _params(get_mock.call_args) |
| 525 | assert p["query"] == "from:steipete -is:retweet" |
| 526 | assert "Grok" not in p["query"] |
| 527 | assert p["start_time"] == "2026-08-10T00:00:00Z" |
| 528 | assert [i["id"] for i in items] == ["XF1"] |
| 529 | assert items[0]["relevance"] >= 0 |
| 530 | |
| 531 | def test_mention_lane_excludes_subjects_own_posts(self, get_mock, fixed_now): |
| 532 | get_mock.return_value = _v2( |
| 533 | [ |
| 534 | _tweet("1", "@steipete nice", author_id="fan"), |
| 535 | _tweet("2", "my own post", author_id="me"), |
| 536 | ], |
| 537 | users=_users(("fan", "fanuser"), ("me", "steipete")), |
| 538 | ) |
| 539 | items = x_api.search_mentions(["steipete"], FROM, TO, topic="Grok 4", count_per=5, token=DUMMY_TOKEN) |
| 540 | p = _params(get_mock.call_args) |
| 541 | assert p["query"] == "@steipete -from:steipete -is:retweet" |
| 542 | assert {i["author_handle"] for i in items} == {"fanuser"} |
| 543 | assert items[0]["id"].startswith("XA") |
| 544 | |
| 545 | def test_invalid_handle_skips_lane_with_receipt(self, get_mock, fixed_now): |
| 546 | items, stderr = _run_capturing_stderr( |
| 547 | lambda: x_api.search_handles(["x OR from:elonmusk"], "t", FROM, TO, token=DUMMY_TOKEN) |
| 548 | ) |
| 549 | assert items == [] |
| 550 | get_mock.assert_not_called() |
| 551 | assert "skip" in stderr.lower() |
| 552 | mentions, _ = _run_capturing_stderr( |
| 553 | lambda: x_api.search_mentions(["bad'; drop", "a" * 16], FROM, TO, token=DUMMY_TOKEN) |
| 554 | ) |
| 555 | assert mentions == [] |
| 556 | get_mock.assert_not_called() |
| 557 | |
| 558 | def test_item_ids_unique_across_handles(self, get_mock, fixed_now): |
| 559 | get_mock.side_effect = [ |
| 560 | _v2([_tweet("1", "a")], users=_users(("u1", "h1"))), |
| 561 | _v2([_tweet("2", "b")], users=_users(("u1", "h2"))), |
| 562 | ] |
| 563 | items = x_api.search_handles(["h1", "h2"], "topic", FROM, TO, token=DUMMY_TOKEN) |
| 564 | ids = [i["id"] for i in items] |
| 565 | assert len(ids) == len(set(ids)) == 2 |
| 566 | |
| 567 | def test_fatal_auth_failure_stops_remaining_handles(self, get_mock, fixed_now): |
| 568 | # Handles fan out on a bounded pool (five in flight), so the handles |
| 569 | # already scheduled beside the failing one may still be called; the |
| 570 | # ones beyond the pool window are not, and nothing leaks. |
| 571 | handles = [f"h{i}" for i in range(1, 8)] |
| 572 | get_mock.side_effect = _http_error(401, body=_LEAKY_BODY) |
| 573 | items, stderr = _run_capturing_stderr( |
| 574 | lambda: x_api.search_handles(handles, "topic", FROM, TO, token=DUMMY_TOKEN) |
| 575 | ) |
| 576 | assert items == [] |
| 577 | assert 1 <= get_mock.call_count <= x_api._MAX_LANE_WORKERS |
| 578 | called = {_params(c)["query"] for c in get_mock.call_args_list} |
| 579 | assert called <= {f"from:{h} -is:retweet" for h in handles} |
| 580 | assert DUMMY_TOKEN not in stderr and ACCOUNT_ID not in stderr |
| 581 | |
| 582 | def test_handle_results_merge_in_handle_order(self, get_mock, fixed_now): |
| 583 | def _by_query(url, headers=None, params=None, **kwargs): |
| 584 | handle = params["query"].split(":")[1].split(" ")[0] |
| 585 | return _v2([_tweet(str(ord(handle[-1])), handle)], users=_users(("u1", handle))) |
| 586 | |
| 587 | get_mock.side_effect = _by_query |
| 588 | handles = ["h3", "h1", "h2"] |
| 589 | items = x_api.search_handles(handles, "topic", FROM, TO, token=DUMMY_TOKEN) |
| 590 | assert [i["author_handle"] for i in items] == handles |
| 591 | assert [i["id"] for i in items] == ["XF1", "XF2", "XF3"] |
| 592 | |
| 593 | def test_transient_failure_continues_to_next_handle(self, get_mock, fixed_now): |
| 594 | get_mock.side_effect = [_http_error(500), _v2([_tweet("2", "b")], users=_users(("u1", "h2")))] |
| 595 | items = x_api.search_handles(["h1", "h2"], "topic", FROM, TO, token=DUMMY_TOKEN) |
| 596 | assert [i["author_handle"] for i in items] == ["h2"] |
| 597 | |
| 598 | def test_no_token_or_handles_returns_empty(self, get_mock): |
| 599 | assert x_api.search_handles(["a"], "t", FROM, TO, token="") == [] |
| 600 | assert x_api.search_handles([], "t", FROM, TO, token=DUMMY_TOKEN) == [] |
| 601 | assert x_api.search_mentions([], FROM, TO, token=DUMMY_TOKEN) == [] |
| 602 | get_mock.assert_not_called() |
| 603 | |
| 604 | |
| 605 | # --------------------------------------------------------------------------- |
| 606 | # Moved helpers (KTD3) |
| 607 | # --------------------------------------------------------------------------- |
| 608 | |
| 609 | |
| 610 | class TestSharedHelpers: |
| 611 | def test_grok_x_imports_helpers_from_x_api(self): |
| 612 | from lib import grok_x |
| 613 | assert grok_x._decode_snowflake is x_api._decode_snowflake |
| 614 | assert grok_x._clean_handle is x_api._clean_handle |
| 615 | assert grok_x._looks_generated is x_api._looks_generated |
| 616 | assert grok_x._HANDLE_RE is x_api._HANDLE_RE |
| 617 | |
| 618 | def test_snowflake_decodes_to_utc_datetime(self): |
| 619 | when = x_api._decode_snowflake("1956158892141441450") |
| 620 | assert when is not None and when.tzinfo is not None |
| 621 | assert when.year == 2025 |
| 622 | assert x_api._decode_snowflake("nope") is None |
| 623 | assert x_api._decode_snowflake("0") is None |
| 624 | |
| 625 | def test_clean_handle_grammar(self): |
| 626 | assert x_api._clean_handle("@steipete") == "steipete" |
| 627 | assert x_api._clean_handle("Peter Steinberger") == "" |
| 628 | assert x_api._clean_handle("a" * 16) == "" |
| 629 | assert x_api._clean_handle("x OR from:elonmusk") == "" |
| 630 | |
| 631 | def test_looks_generated_flags_uniform_runs(self): |
| 632 | assert x_api._looks_generated(["100", "200", "300", "400"]) is True |
| 633 | assert x_api._looks_generated(["100", "250", "300", "900"]) is False |
| 634 | assert x_api._looks_generated(["1", "2"]) is False |
| 635 | |
| 636 | |
| 637 | # --------------------------------------------------------------------------- |
| 638 | # Pipeline wiring (R7): lane selection and failover line |
| 639 | # --------------------------------------------------------------------------- |
| 640 | |
| 641 | |
| 642 | def _make_source_item(source, item_id, url, author=None, body=""): |
| 643 | return schema.SourceItem( |
| 644 | item_id=item_id, source=source, title=f"Item {item_id}", body=body, url=url, author=author, |
| 645 | ) |
| 646 | |
| 647 | |
| 648 | def _make_plan(topic): |
| 649 | return schema.QueryPlan( |
| 650 | intent="exploration", freshness_mode="balanced_recent", cluster_mode="topic", |
| 651 | raw_topic=topic, |
| 652 | subqueries=[schema.SubQuery( |
| 653 | label="primary", search_query=topic, |
| 654 | ranking_query=f"What recent evidence matters for {topic}?", sources=["x"], |
| 655 | )], |
| 656 | source_weights={"x": 1.0}, |
| 657 | ) |
| 658 | |
| 659 | |
| 660 | class TestPipelineWiring: |
| 661 | def test_chain_xapi_yields_primary_xapi_and_runs_both_lanes(self): |
| 662 | bundle = schema.RetrievalBundle() |
| 663 | bundle.items_by_source["x"] = [ |
| 664 | _make_source_item("x", "X1", "https://x.com/analyst1/status/1", author="analyst1", body="AI safety analysis"), |
| 665 | _make_source_item("x", "X2", "https://x.com/analyst1/status/2", author="analyst1", body="AI safety research"), |
| 666 | ] |
| 667 | from_items = [{ |
| 668 | "id": "XF1", "text": "from analyst1", "url": "https://x.com/analyst1/status/777", |
| 669 | "author_handle": "analyst1", "date": "2026-03-15", |
| 670 | "engagement": {"likes": 30}, "relevance": 0.8, "why_relevant": "", |
| 671 | }] |
| 672 | runtime = schema.ProviderRuntime( |
| 673 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", x_search_backend=None, |
| 674 | ) |
| 675 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi"]), \ |
| 676 | mock.patch("lib.entity_extract.extract_entities", |
| 677 | return_value={"x_handles": ["analyst1"], "x_hashtags": [], "reddit_subreddits": []}), \ |
| 678 | mock.patch("lib.x_api.search_handles", return_value=from_items) as from_lane, \ |
| 679 | mock.patch("lib.x_api.search_mentions", return_value=[]) as about_lane, \ |
| 680 | mock.patch("lib.bird_x.search_handles") as bird_lane, \ |
| 681 | mock.patch("lib.xquik.search_handles") as xquik_lane: |
| 682 | pipeline._run_supplemental_searches( |
| 683 | topic="AI safety", bundle=bundle, plan=_make_plan("AI safety"), |
| 684 | config={"X_BEARER_TOKEN": DUMMY_TOKEN}, depth="default", |
| 685 | date_range=("2026-02-15", "2026-03-17"), runtime=runtime, mock=False, |
| 686 | rate_limited_sources=set(), rate_limit_lock=threading.Lock(), |
| 687 | ) |
| 688 | from_lane.assert_called_once() |
| 689 | about_lane.assert_called_once() |
| 690 | assert from_lane.call_args.kwargs["token"] == DUMMY_TOKEN |
| 691 | assert about_lane.call_args.kwargs["token"] == DUMMY_TOKEN |
| 692 | bird_lane.assert_not_called() |
| 693 | xquik_lane.assert_not_called() |
| 694 | x_urls = {item.url for item in bundle.items_by_source.get("x", [])} |
| 695 | assert "https://x.com/analyst1/status/777" in x_urls |
| 696 | |
| 697 | def test_xapi_lanes_share_one_deadline(self): |
| 698 | """Explicit-from, extracted-from, about, and related lanes must draw |
| 699 | on one wall-clock budget, never a fresh 90s each (review finding).""" |
| 700 | bundle = schema.RetrievalBundle() |
| 701 | bundle.items_by_source["x"] = [ |
| 702 | _make_source_item("x", "X1", "https://x.com/analyst1/status/1", author="analyst1", body="AI safety analysis"), |
| 703 | ] |
| 704 | runtime = schema.ProviderRuntime( |
| 705 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", x_search_backend=None, |
| 706 | ) |
| 707 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi"]), \ |
| 708 | mock.patch("lib.entity_extract.extract_entities", |
| 709 | return_value={"x_handles": ["analyst1"], "x_hashtags": [], "reddit_subreddits": []}), \ |
| 710 | mock.patch("lib.x_api.search_handles", return_value=[]) as from_lane, \ |
| 711 | mock.patch("lib.x_api.search_mentions", return_value=[]) as about_lane: |
| 712 | pipeline._run_supplemental_searches( |
| 713 | topic="AI safety", bundle=bundle, plan=_make_plan("AI safety"), |
| 714 | config={"X_BEARER_TOKEN": DUMMY_TOKEN}, depth="default", |
| 715 | date_range=("2026-02-15", "2026-03-17"), runtime=runtime, mock=False, |
| 716 | rate_limited_sources=set(), rate_limit_lock=threading.Lock(), |
| 717 | x_handle="steipete", x_related=["peer1"], |
| 718 | ) |
| 719 | calls = from_lane.call_args_list + about_lane.call_args_list |
| 720 | assert len(calls) >= 3, "explicit from, extracted from, about, related" |
| 721 | deadlines = {c.kwargs.get("deadline") for c in calls} |
| 722 | assert len(deadlines) == 1 and None not in deadlines, deadlines |
| 723 | |
| 724 | def test_xapi_lane_deadline_stop_reaches_report_warnings(self): |
| 725 | """A lane cut short by the deadline is reported as partial coverage, |
| 726 | not presented as complete (review finding).""" |
| 727 | bundle = schema.RetrievalBundle() |
| 728 | bundle.items_by_source["x"] = [ |
| 729 | _make_source_item("x", "X1", "https://x.com/analyst1/status/1", author="analyst1", body="AI safety analysis"), |
| 730 | ] |
| 731 | runtime = schema.ProviderRuntime( |
| 732 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", x_search_backend=None, |
| 733 | ) |
| 734 | |
| 735 | def cut_short(*args, **kwargs): |
| 736 | kwargs["warnings"].append(x_api.DEADLINE_DETAIL) |
| 737 | return [] |
| 738 | |
| 739 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi"]), \ |
| 740 | mock.patch("lib.entity_extract.extract_entities", |
| 741 | return_value={"x_handles": [], "x_hashtags": [], "reddit_subreddits": []}), \ |
| 742 | mock.patch("lib.x_api.search_handles", side_effect=cut_short), \ |
| 743 | mock.patch("lib.x_api.search_mentions", side_effect=cut_short): |
| 744 | pipeline._run_supplemental_searches( |
| 745 | topic="AI safety", bundle=bundle, plan=_make_plan("AI safety"), |
| 746 | config={"X_BEARER_TOKEN": DUMMY_TOKEN}, depth="default", |
| 747 | date_range=("2026-02-15", "2026-03-17"), runtime=runtime, mock=False, |
| 748 | rate_limited_sources=set(), rate_limit_lock=threading.Lock(), |
| 749 | x_handle="steipete", |
| 750 | ) |
| 751 | receipts = [w for w in bundle.artifacts.get("x_partial_coverage", []) if x_api.DEADLINE_DETAIL in w] |
| 752 | assert receipts == [f"X handle lanes: {x_api.DEADLINE_DETAIL}"], bundle.artifacts.get("x_partial_coverage") |
| 753 | |
| 754 | def test_lane_search_past_a_shared_deadline_sends_no_request(self, get_mock, fixed_now, monkeypatch): |
| 755 | monkeypatch.setattr(x_api.time, "monotonic", lambda: 1000.0) |
| 756 | notes: list[str] = [] |
| 757 | assert x_api.search_handles(["steipete", "peer1"], "t", FROM, TO, token=DUMMY_TOKEN, deadline=999.0, warnings=notes) == [] |
| 758 | assert x_api.search_mentions(["steipete"], FROM, TO, token=DUMMY_TOKEN, deadline=999.0, warnings=notes) == [] |
| 759 | get_mock.assert_not_called() |
| 760 | assert notes == [x_api.DEADLINE_DETAIL], "one receipt per lane call, deduped across handles" |
| 761 | get_mock.return_value = _v2([_tweet("1", "hi")], users=_users(("u1", "steipete"))) |
| 762 | assert len(x_api.search_handles(["steipete"], "t", FROM, TO, token=DUMMY_TOKEN, deadline=1001.0)) == 1 |
| 763 | |
| 764 | def test_all_backends_failed_keeps_the_payment_required_state(self): |
| 765 | """xapi's 402 must not be masked by a later backend's generic failure: |
| 766 | the outcome that reaches doctor says top up, not re-authenticate.""" |
| 767 | plan = { |
| 768 | "intent": "general", "freshness_mode": "balanced_recent", "cluster_mode": "story", |
| 769 | "subqueries": [{ |
| 770 | "label": "primary", "search_query": "topic", |
| 771 | "ranking_query": "What are people saying about topic?", "sources": ["x"], |
| 772 | }], |
| 773 | "source_weights": {"x": 1.0}, |
| 774 | } |
| 775 | answers = iter([ |
| 776 | ([], x_api.ERR_PAYMENT_REQUIRED), |
| 777 | ([], "request failed (HTTPError)"), |
| 778 | ]) |
| 779 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi", "xai"]), \ |
| 780 | mock.patch("lib.pipeline._fetch_x_backend", side_effect=lambda *a, **k: next(answers)): |
| 781 | report = pipeline.run( |
| 782 | topic="topic", config={"X_BEARER_TOKEN": DUMMY_TOKEN, "XAI_API_KEY": "dummy-xai"}, |
| 783 | depth="quick", requested_sources=["x"], mock=False, |
| 784 | external_plan=plan, web_backend="none", save_dir="", |
| 785 | ) |
| 786 | assert report.source_status["x"].state == health.PAYMENT_REQUIRED |
| 787 | assert "payment required" in report.source_status["x"].detail |
| 788 | |
| 789 | def test_fetch_x_backend_registers_xapi(self, get_mock, fixed_now): |
| 790 | get_mock.return_value = _v2([_tweet("1", "hello topic")], users=_users(("u1", "a"))) |
| 791 | items, err = pipeline._fetch_x_backend( |
| 792 | "xapi", "topic", FROM, TO, "quick", {"X_BEARER_TOKEN": DUMMY_TOKEN}, |
| 793 | ) |
| 794 | assert err == "" |
| 795 | assert [i["id"] for i in items] == ["XAPI1"] |
| 796 | assert get_mock.call_args.kwargs["headers"]["Authorization"] == f"Bearer {DUMMY_TOKEN}" |
| 797 | |
| 798 | def test_fetch_x_backend_surfaces_truncation_warning(self, get_mock, fixed_now): |
| 799 | get_mock.side_effect = [ |
| 800 | _http_error(403, body="client-not-enrolled"), |
| 801 | _v2([_tweet("1", "hello topic", created_at="2026-09-05T00:00:00Z")]), |
| 802 | ] |
| 803 | warnings: list[str] = [] |
| 804 | items, err = pipeline._fetch_x_backend( |
| 805 | "xapi", "topic", FROM, TO, "quick", {"X_BEARER_TOKEN": DUMMY_TOKEN}, warnings=warnings, |
| 806 | ) |
| 807 | assert err == "" and len(items) == 1 |
| 808 | assert warnings and "window truncated to 7 days" in warnings[0] |
| 809 | |
| 810 | def test_truncation_receipt_reaches_the_stream_artifact(self, get_mock, fixed_now): |
| 811 | get_mock.side_effect = [ |
| 812 | _http_error(403, body="client-not-enrolled"), |
| 813 | _v2([_tweet("1", "hello topic", created_at="2026-09-05T00:00:00Z")]), |
| 814 | ] |
| 815 | sq = schema.SubQuery(label="primary", search_query="q", ranking_query="q?", sources=["x"]) |
| 816 | runtime = schema.ProviderRuntime( |
| 817 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", x_search_backend=None, |
| 818 | ) |
| 819 | |
| 820 | def run(): |
| 821 | with mock.patch("lib.env.x_backend_chain", return_value=["xapi"]): |
| 822 | return pipeline._retrieve_stream( |
| 823 | topic="q", subquery=sq, source="x", config={"X_BEARER_TOKEN": DUMMY_TOKEN}, |
| 824 | depth="quick", date_range=(FROM, TO), runtime=runtime, mock=False, |
| 825 | ) |
| 826 | |
| 827 | (items, artifact), stderr = _run_capturing_stderr(run) |
| 828 | assert len(items) == 1 |
| 829 | assert artifact["x_receipts"] == ["X: xapi window truncated to 7 days"] |
| 830 | assert "window truncated to 7 days" in stderr |
| 831 | |
| 832 | def test_failover_stderr_line_names_xapi_and_credits(self): |
| 833 | sq = schema.SubQuery(label="primary", search_query="q", ranking_query="q?", sources=["x"]) |
| 834 | runtime = schema.ProviderRuntime( |
| 835 | reasoning_provider="mock", planner_model="mock", rerank_model="mock", x_search_backend=None, |
| 836 | ) |
| 837 | good = [{"id": "XAPI1", "text": "t", "url": "https://x.com/a/status/1", "author_handle": "a", |
| 838 | "date": "2026-08-20", "engagement": None, "relevance": 0.5, "why_relevant": ""}] |
| 839 | |
| 840 | def fetch(backend, *_args, **_kwargs): |
| 841 | return (good, "") if backend == "xapi" else ([], "") |
| 842 | |
| 843 | def run(): |
| 844 | with mock.patch("lib.env.x_backend_chain", return_value=["xai", "xapi"]), \ |
| 845 | mock.patch("lib.pipeline._fetch_x_backend", side_effect=fetch): |
| 846 | return pipeline._retrieve_stream( |
| 847 | topic="q", subquery=sq, source="x", config={"X_BEARER_TOKEN": DUMMY_TOKEN}, |
| 848 | depth="quick", date_range=(FROM, TO), runtime=runtime, mock=False, |
| 849 | ) |
| 850 | |
| 851 | (items, _artifact), stderr = _run_capturing_stderr(run) |
| 852 | assert len(items) == 1 |
| 853 | assert "used fallback 'xapi'" in stderr |
| 854 | assert "X API credits" in stderr |
| 855 | |
| 856 | |
| 857 | # --------------------------------------------------------------------------- |
| 858 | # Fixture redaction at both seams (R8) |
| 859 | # --------------------------------------------------------------------------- |
| 860 | |
| 861 | |
| 862 | def _urlopen_response(body: str): |
| 863 | response = MagicMock() |
| 864 | response.__enter__.return_value = response |
| 865 | response.__exit__.return_value = False |
| 866 | response.read.return_value = body.encode("utf-8") |
| 867 | response.status = 200 |
| 868 | return response |
| 869 | |
| 870 | |
| 871 | def test_fixture_redacts_bearer_loaded_only_from_env_file(tmp_path, monkeypatch): |
| 872 | env_file = tmp_path / ".env" |
| 873 | env_file.write_text(f"X_BEARER_TOKEN={DUMMY_TOKEN}\n", encoding="utf-8") |
| 874 | env_file.chmod(0o600) |
| 875 | monkeypatch.setattr(env, "CONFIG_FILE", env_file) |
| 876 | monkeypatch.delenv("X_BEARER_TOKEN", raising=False) |
| 877 | monkeypatch.setenv("LAST30DAYS_SKIP_KEYCHAIN", "1") |
| 878 | monkeypatch.setattr(env, "_load_pass", lambda *a, **k: {}) |
| 879 | monkeypatch.setattr(env, "_find_project_env", lambda: None) |
| 880 | monkeypatch.setattr( |
| 881 | http.urllib.request, "urlopen", |
| 882 | lambda *_a, **_k: _urlopen_response(json.dumps({"detail": f"token {DUMMY_TOKEN} rejected"})), |
| 883 | ) |
| 884 | fixture_dir = tmp_path / "fixture" |
| 885 | |
| 886 | with http.recording_requests(fixture_dir): |
| 887 | config = env.get_config() |
| 888 | assert config["X_BEARER_TOKEN"] == DUMMY_TOKEN |
| 889 | http.get( |
| 890 | "https://api.x.com/2/tweets/search/all", |
| 891 | headers={"Authorization": f"Bearer {DUMMY_TOKEN}"}, |
| 892 | params={"query": "topic"}, |
| 893 | ) |
| 894 | http.fixture_source_record( |
| 895 | {"source": "x", "topic": "t", "search_query": "t", "date_range": [FROM, TO], "depth": "quick"}, |
| 896 | [[{"text": f"echo {DUMMY_TOKEN}"}], {}], |
| 897 | ) |
| 898 | |
| 899 | payload = json.loads((fixture_dir / "http.json").read_text(encoding="utf-8")) |
| 900 | text = json.dumps(payload) |
| 901 | assert DUMMY_TOKEN not in text |
| 902 | assert "<redacted>" in json.dumps(payload["exchanges"]) |
| 903 | assert "<redacted>" in json.dumps(payload["source_exchanges"]) |
| 904 | |
| 905 | |
| 906 | def test_http_seam_redacts_bare_bearer_without_any_config(tmp_path, monkeypatch): |
| 907 | monkeypatch.delenv("X_BEARER_TOKEN", raising=False) |
| 908 | monkeypatch.setattr( |
| 909 | http.urllib.request, "urlopen", |
| 910 | lambda *_a, **_k: _urlopen_response(json.dumps({"echo": DUMMY_TOKEN})), |
| 911 | ) |
| 912 | fixture_dir = tmp_path / "fixture" |
| 913 | with http.recording_requests(fixture_dir): |
| 914 | http.get("https://api.x.com/2/tweets/search/all", headers={"Authorization": f"Bearer {DUMMY_TOKEN}"}) |
| 915 | text = (fixture_dir / "http.json").read_text(encoding="utf-8") |
| 916 | assert DUMMY_TOKEN not in text |
| 917 | assert '"echo": "<redacted>"' in text |
| 918 | |
| 919 | |
| 920 | # --------------------------------------------------------------------------- |
| 921 | # xurl fixed strings (same rule as xapi) |
| 922 | # --------------------------------------------------------------------------- |
| 923 | |
| 924 | |
| 925 | class TestXurlFixedStrings: |
| 926 | def _run(self, returncode, stderr="", stdout=""): |
| 927 | completed = mock.Mock(returncode=returncode, stdout=stdout, stderr=stderr) |
| 928 | with mock.patch("subprocess.run", return_value=completed): |
| 929 | return xurl_x.search_x("test") |
| 930 | |
| 931 | def test_rate_limited_stderr_with_token_maps_to_fixed_string(self): |
| 932 | result = self._run(1, stderr=f"429 rate limit exceeded for {DUMMY_TOKEN} ({ACCOUNT_ID})") |
| 933 | assert result["error"] == xurl_x.ERR_RATE_LIMITED |
| 934 | assert DUMMY_TOKEN not in result["error"] and ACCOUNT_ID not in result["error"] |
| 935 | assert http.classify_failure(message=result["error"]) == health.RATE_LIMITED |
| 936 | |
| 937 | @pytest.mark.parametrize("stderr,expected,state", [ |
| 938 | ("401 Unauthorized", "ERR_UNAUTHORIZED", health.AUTH_FAILED), |
| 939 | ("403 Forbidden: client-not-enrolled", "ERR_FORBIDDEN", health.AUTH_FAILED), |
| 940 | ("402 Payment Required", "ERR_PAYMENT_REQUIRED", health.PAYMENT_REQUIRED), |
| 941 | ("something else entirely", "ERR_FAILED", health.ERROR), |
| 942 | ]) |
| 943 | def test_status_words_map_to_fixed_strings(self, stderr, expected, state): |
| 944 | result = self._run(1, stderr=f"{stderr} {DUMMY_TOKEN}") |
| 945 | assert result["error"] == getattr(xurl_x, expected) |
| 946 | assert DUMMY_TOKEN not in result["error"] |
| 947 | assert http.classify_failure(message=result["error"]) == state |
| 948 | |
| 949 | def test_invalid_json_is_fixed_string(self): |
| 950 | result = self._run(0, stdout=f"NOT JSON {DUMMY_TOKEN}") |
| 951 | assert result["error"] == xurl_x.ERR_INVALID_JSON |
| 952 | assert DUMMY_TOKEN not in result["error"] |
| 953 |