| 1 | """grok_x: X retrieval via the Grok CLI. |
| 2 | |
| 3 | The failure modes pinned here were all measured against grok CLI 0.2.118 on |
| 4 | 2026-08-13, not hypothesized. Retrieval is performed by a language model |
| 5 | rather than an API client, so the module's job is as much rejecting confident |
| 6 | fabrication as it is parsing. |
| 7 | """ |
| 8 | |
| 9 | import json |
| 10 | import re |
| 11 | import subprocess |
| 12 | |
| 13 | import pytest |
| 14 | |
| 15 | from lib import grok_x |
| 16 | |
| 17 | |
| 18 | @pytest.fixture(autouse=True) |
| 19 | def _reset(): |
| 20 | grok_x.clear_availability_cache() |
| 21 | yield |
| 22 | grok_x.clear_availability_cache() |
| 23 | |
| 24 | |
| 25 | def _block(post_id, handle="steipete", created="Wed, 12 Aug 2026 15:55:18 GMT", |
| 26 | likes=1462, text="cli was a year ago."): |
| 27 | return ( |
| 28 | f"id: {post_id}\n" |
| 29 | f"handle: {handle}\n" |
| 30 | f"created_at: {created}\n" |
| 31 | f"likes: {likes}\n" |
| 32 | f"reposts: 48\nreplies: 95\nquotes: 20\n" |
| 33 | f"text: {text}\n" |
| 34 | ) |
| 35 | |
| 36 | |
| 37 | WINDOW = ("2026-07-14", "2026-08-13") |
| 38 | |
| 39 | |
| 40 | # --- provenance: the primary validity test -------------------------------- |
| 41 | |
| 42 | def test_in_window_post_is_parsed(): |
| 43 | items = grok_x.parse_x_response( |
| 44 | {"text": _block("2087568620465607078")}, "steipete", *WINDOW |
| 45 | ) |
| 46 | assert len(items) == 1 |
| 47 | item = items[0] |
| 48 | assert item["author_handle"] == "steipete" |
| 49 | assert item["url"] == "https://x.com/steipete/status/2087568620465607078" |
| 50 | assert item["date"] == "2026-08-12" |
| 51 | assert item["engagement"]["likes"] == 1462 |
| 52 | |
| 53 | |
| 54 | def test_out_of_window_ids_are_rejected(): |
| 55 | """The measured fabrication: a since:2026-07-14 request answered with 2025 |
| 56 | posts recalled from training data. Handle, id format, text and engagement |
| 57 | all looked correct; only the decoded timestamp exposed it.""" |
| 58 | text = "".join( |
| 59 | _block(pid, created="Fri, 15 Aug 2025 00:59:58 GMT") |
| 60 | for pid in ("1956158892141441450", "1955681419900834272") |
| 61 | ) |
| 62 | assert grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) == [] |
| 63 | |
| 64 | |
| 65 | def test_uniform_id_sequence_is_rejected(): |
| 66 | """A generated series: real ranked results are not evenly spaced.""" |
| 67 | base = 2080000000000000000 |
| 68 | step = 715000000000000 |
| 69 | text = "".join(_block(str(base + i * step)) for i in range(6)) |
| 70 | assert grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) == [] |
| 71 | |
| 72 | |
| 73 | def test_placeholder_handle_is_rejected(): |
| 74 | text = _block("2087568620465607078", handle="unknown", likes=0) |
| 75 | assert grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) == [] |
| 76 | |
| 77 | |
| 78 | def test_self_reported_non_execution_is_rejected(): |
| 79 | text = _block( |
| 80 | "2087568620465607078", |
| 81 | text="Requested x_keyword_search was not executed in this turn", |
| 82 | ) |
| 83 | assert grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) == [] |
| 84 | |
| 85 | |
| 86 | def test_item_without_usable_id_is_dropped(): |
| 87 | text = "handle: steipete\nlikes: 10\ntext: no id here\n" |
| 88 | assert grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) == [] |
| 89 | |
| 90 | |
| 91 | def test_mixed_response_keeps_only_in_window_posts(): |
| 92 | text = _block("2087568620465607078") + _block( |
| 93 | "1956158892141441450", created="Fri, 15 Aug 2025 00:59:58 GMT" |
| 94 | ) |
| 95 | items = grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) |
| 96 | assert [i["url"].rsplit("/", 1)[-1] for i in items] == ["2087568620465607078"] |
| 97 | |
| 98 | |
| 99 | def test_duplicate_post_ids_are_deduped(): |
| 100 | text = _block("2087568620465607078") * 2 |
| 101 | assert len(grok_x.parse_x_response({"text": text}, "steipete", *WINDOW)) == 1 |
| 102 | |
| 103 | |
| 104 | # --- parsing resilience ---------------------------------------------------- |
| 105 | |
| 106 | def test_narration_around_blocks_is_tolerated(): |
| 107 | text = ( |
| 108 | "I'll call x_keyword_search now.\n\n" |
| 109 | "Here are the posts I found:\n\n" |
| 110 | + _block("2087568620465607078") |
| 111 | + "\nThat's everything the tool returned.\n" |
| 112 | ) |
| 113 | assert len(grok_x.parse_x_response({"text": text}, "steipete", *WINDOW)) == 1 |
| 114 | |
| 115 | |
| 116 | def test_json_array_with_url_fields_is_parsed(): |
| 117 | """Grok CLI 1.0.5 emits a JSON array of {url, text, likes} instead of |
| 118 | the prompted id:/handle: field blocks (#1051).""" |
| 119 | text = json.dumps([ |
| 120 | { |
| 121 | "url": "https://x.com/steipete/status/2087568620465607078", |
| 122 | "text": "cli was a year ago.", |
| 123 | "likes": 2041, |
| 124 | } |
| 125 | ]) |
| 126 | items = grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) |
| 127 | assert len(items) == 1 |
| 128 | assert items[0]["author_handle"] == "steipete" |
| 129 | assert items[0]["url"].endswith("/2087568620465607078") |
| 130 | assert items[0]["engagement"]["likes"] == 2041 |
| 131 | |
| 132 | |
| 133 | def test_fenced_json_preamble_is_parsed(): |
| 134 | """Without --output-format json the CLI narrates, then fences the array.""" |
| 135 | text = ( |
| 136 | "I'll search X for Seedance posts from the last 7 days and return " |
| 137 | "only the JSON array.\n" |
| 138 | "Checking the X search workflow first.\n" |
| 139 | "```json\n" |
| 140 | + json.dumps([ |
| 141 | { |
| 142 | "url": "https://x.com/steipete/status/2087568620465607078", |
| 143 | "text": "cli was a year ago.", |
| 144 | "likes": 1462, |
| 145 | } |
| 146 | ]) |
| 147 | + "\n```\n" |
| 148 | ) |
| 149 | items = grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) |
| 150 | assert len(items) == 1 |
| 151 | assert items[0]["author_handle"] == "steipete" |
| 152 | |
| 153 | |
| 154 | def test_markdown_decorated_fields_are_parsed(): |
| 155 | text = ( |
| 156 | "- **id:** 2087568620465607078\n" |
| 157 | "- **handle:** @steipete\n" |
| 158 | "- **created_at:** Wed, 12 Aug 2026 15:55:18 GMT\n" |
| 159 | "- **likes:** 1,462\n" |
| 160 | "- **text:** cli was a year ago.\n" |
| 161 | ) |
| 162 | items = grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) |
| 163 | assert len(items) == 1 |
| 164 | assert items[0]["engagement"]["likes"] == 1462 |
| 165 | |
| 166 | |
| 167 | def test_error_response_returns_empty_not_raises(): |
| 168 | assert grok_x.parse_x_response({"error": "boom"}, "t", *WINDOW) == [] |
| 169 | |
| 170 | |
| 171 | def test_non_dict_response_returns_empty(): |
| 172 | assert grok_x.parse_x_response(None, "t", *WINDOW) == [] |
| 173 | |
| 174 | |
| 175 | # --- invocation contract --------------------------------------------------- |
| 176 | |
| 177 | def test_invocation_omits_json_schema_and_tools(monkeypatch): |
| 178 | """Both flags degrade or suppress the tool call; neither may be passed.""" |
| 179 | seen = {} |
| 180 | |
| 181 | def fake_run(cmd, **kwargs): |
| 182 | seen["cmd"] = cmd |
| 183 | seen["kwargs"] = kwargs |
| 184 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 185 | |
| 186 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 187 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 188 | grok_x.search_x("steipete", *WINDOW) |
| 189 | assert "--json-schema" not in seen["cmd"] |
| 190 | assert "--tools" not in seen["cmd"] |
| 191 | assert "--permission-mode" in seen["cmd"] |
| 192 | # Grok 1.0.5 narrates before the payload unless stdout is JSON (#1051). |
| 193 | # --json-schema is still forbidden: constrained decoding skips the tool. |
| 194 | assert "--output-format" in seen["cmd"] |
| 195 | fmt_idx = seen["cmd"].index("--output-format") |
| 196 | assert seen["cmd"][fmt_idx + 1] == "json" |
| 197 | |
| 198 | |
| 199 | def test_subprocess_runs_in_an_isolated_empty_directory(monkeypatch): |
| 200 | """The child has tool permissions bypassed and its context carries |
| 201 | untrusted post text, so it must not act in the user's repository.""" |
| 202 | import os |
| 203 | seen = {} |
| 204 | |
| 205 | def fake_run(cmd, **kwargs): |
| 206 | seen["cwd"] = kwargs.get("cwd") |
| 207 | seen["existed"] = os.path.isdir(kwargs.get("cwd") or "") |
| 208 | seen["entries"] = os.listdir(kwargs.get("cwd")) if seen["existed"] else None |
| 209 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 210 | |
| 211 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 212 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 213 | grok_x.search_x("steipete", *WINDOW) |
| 214 | assert seen["cwd"] and seen["cwd"] != os.getcwd() |
| 215 | # Isolated and near-empty: the only entry is the throwaway HOME staged for |
| 216 | # the child, never the user's checkout. |
| 217 | assert seen["existed"] and seen["entries"] == ["home"] |
| 218 | |
| 219 | |
| 220 | def test_subprocess_environment_is_minimal(monkeypatch): |
| 221 | monkeypatch.setenv("XAI_API_KEY", "dummy-key") |
| 222 | monkeypatch.setenv("AUTH_TOKEN", "dummy-token") |
| 223 | seen = {} |
| 224 | |
| 225 | def fake_run(cmd, **kwargs): |
| 226 | seen["env"] = kwargs.get("env") |
| 227 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 228 | |
| 229 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 230 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 231 | grok_x.search_x("steipete", *WINDOW) |
| 232 | assert "XAI_API_KEY" not in seen["env"] |
| 233 | assert "AUTH_TOKEN" not in seen["env"] |
| 234 | assert "PATH" in seen["env"] |
| 235 | |
| 236 | |
| 237 | def test_resolved_binary_path_is_used_not_bare_name(monkeypatch): |
| 238 | """shutil.which and subprocess.run resolve a bare name differently on |
| 239 | Windows, so the resolved path must be passed.""" |
| 240 | seen = {} |
| 241 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/opt/custom/grok") |
| 242 | |
| 243 | def fake_run(cmd, **kwargs): |
| 244 | seen["cmd"] = cmd |
| 245 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 246 | |
| 247 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 248 | grok_x.search_x("steipete", *WINDOW) |
| 249 | assert seen["cmd"][0] == "/opt/custom/grok" |
| 250 | |
| 251 | |
| 252 | def test_subprocess_decodes_stdout_as_utf8_not_locale(monkeypatch): |
| 253 | """text=True with no explicit encoding decodes with the locale codec |
| 254 | (cp1252 on Windows). X post text is not cp1252, so a bare emoji or |
| 255 | smart quote in the CLI's stdout would otherwise crash the decode and |
| 256 | surface as "no items parsed" instead of a real error.""" |
| 257 | seen = {} |
| 258 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 259 | |
| 260 | def fake_run(cmd, **kwargs): |
| 261 | seen["kwargs"] = kwargs |
| 262 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 263 | |
| 264 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 265 | grok_x.search_x("steipete", *WINDOW) |
| 266 | assert seen["kwargs"].get("encoding") == "utf-8" |
| 267 | assert seen["kwargs"].get("errors") == "replace" |
| 268 | |
| 269 | |
| 270 | def test_missing_binary_returns_error_not_raises(monkeypatch): |
| 271 | monkeypatch.setattr(grok_x, "binary_path", lambda: None) |
| 272 | result = grok_x.search_x("steipete", *WINDOW) |
| 273 | assert result["items"] == [] and result["error"] |
| 274 | |
| 275 | |
| 276 | def test_timeout_returns_error_not_raises(monkeypatch): |
| 277 | def fake_run(cmd, **kwargs): |
| 278 | raise subprocess.TimeoutExpired(cmd, 1) |
| 279 | |
| 280 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 281 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 282 | result = grok_x.search_x("steipete", *WINDOW) |
| 283 | assert result["items"] == [] and "timed out" in result["error"] |
| 284 | |
| 285 | |
| 286 | def test_non_execution_is_retried(monkeypatch): |
| 287 | """A response that fails provenance is a retryable non-execution, not a |
| 288 | thin result -- the measured rate made single-shot unreliable. |
| 289 | |
| 290 | Asserted against _run_query directly: search_x additionally fans out across |
| 291 | query variants to reach the depth target, which would confound a call count. |
| 292 | """ |
| 293 | calls = {"n": 0} |
| 294 | |
| 295 | def fake_run(cmd, **kwargs): |
| 296 | calls["n"] += 1 |
| 297 | body = ( |
| 298 | _block("1956158892141441450", created="Fri, 15 Aug 2025 00:59:58 GMT") |
| 299 | if calls["n"] == 1 else _block("2087568620465607078") |
| 300 | ) |
| 301 | return subprocess.CompletedProcess(cmd, 0, body, "") |
| 302 | |
| 303 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 304 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 305 | items, error, auth_revoked = grok_x._run_query("steipete", *WINDOW) |
| 306 | assert calls["n"] == 2, "a provenance rejection must be retried once" |
| 307 | assert not auth_revoked |
| 308 | assert len(items) == 1 and not error |
| 309 | |
| 310 | |
| 311 | # --- auth surfaces --------------------------------------------------------- |
| 312 | |
| 313 | def test_stored_auth_status_makes_no_subprocess_or_network(monkeypatch): |
| 314 | """This is the doctor path; the whole-doctor test patches these to raise.""" |
| 315 | def boom(*a, **k): |
| 316 | raise AssertionError("doctor path must not spawn a process or hit the network") |
| 317 | |
| 318 | monkeypatch.setattr(grok_x.subprocess, "run", boom) |
| 319 | monkeypatch.setattr(grok_x.subprocess, "Popen", boom) |
| 320 | grok_x.stored_auth_status() |
| 321 | |
| 322 | |
| 323 | def test_stored_auth_status_reports_missing_store(monkeypatch, tmp_path): |
| 324 | monkeypatch.setattr(grok_x, "token_store_path", lambda: tmp_path / "nope.json") |
| 325 | status, detail, expires_at = grok_x.stored_auth_status() |
| 326 | assert status == grok_x.AUTH_MISSING and "nope.json" in detail |
| 327 | assert expires_at is None |
| 328 | |
| 329 | |
| 330 | def test_stored_auth_status_detects_credentials(monkeypatch, tmp_path): |
| 331 | store = tmp_path / "auth.json" |
| 332 | store.write_text('{"iss": {"auth_mode": "oidc", "refresh_token": "dummy-token"}}') |
| 333 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 334 | assert grok_x.stored_auth_status()[0] == grok_x.AUTH_OK |
| 335 | |
| 336 | |
| 337 | def test_stored_auth_status_never_echoes_store_contents(monkeypatch, tmp_path): |
| 338 | """doctor output gets pasted into issue reports.""" |
| 339 | store = tmp_path / "auth.json" |
| 340 | store.write_text('{"key": "SUPER-SECRET-VALUE", "refresh_token": "ALSO-SECRET"}') |
| 341 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 342 | _, detail, _ = grok_x.stored_auth_status() |
| 343 | assert "SUPER-SECRET-VALUE" not in detail and "ALSO-SECRET" not in detail |
| 344 | |
| 345 | |
| 346 | def test_unreadable_store_is_an_error(monkeypatch, tmp_path): |
| 347 | store = tmp_path / "auth.json" |
| 348 | store.write_text("{}") |
| 349 | |
| 350 | def boom(*a, **k): |
| 351 | raise OSError("permission denied") |
| 352 | |
| 353 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 354 | monkeypatch.setattr(type(store), "read_text", boom, raising=False) |
| 355 | assert grok_x.stored_auth_status()[0] == grok_x.AUTH_ERROR |
| 356 | |
| 357 | |
| 358 | # --- expires_at parsing ----------------------------------------------------- |
| 359 | |
| 360 | def test_stored_auth_status_future_expires_at_is_ok(monkeypatch, tmp_path): |
| 361 | """Credentials with expires_at in the future report AUTH_OK.""" |
| 362 | from datetime import datetime, timezone, timedelta |
| 363 | future = (datetime.now(timezone.utc) + timedelta(hours=2)).isoformat() |
| 364 | store = tmp_path / "auth.json" |
| 365 | store.write_text(f'{{"iss": {{"refresh_token": "tok", "expires_at": "{future}"}}}}') |
| 366 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 367 | status, detail, expires_at = grok_x.stored_auth_status() |
| 368 | assert status == grok_x.AUTH_OK |
| 369 | assert expires_at is not None |
| 370 | |
| 371 | |
| 372 | def test_stored_auth_status_past_expires_at_is_expired(monkeypatch, tmp_path): |
| 373 | """Credentials with expires_at in the past report AUTH_EXPIRED, not AUTH_OK.""" |
| 374 | from datetime import datetime, timezone, timedelta |
| 375 | past = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat() |
| 376 | store = tmp_path / "auth.json" |
| 377 | store.write_text(f'{{"iss": {{"refresh_token": "tok", "expires_at": "{past}"}}}}') |
| 378 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 379 | status, detail, expires_at = grok_x.stored_auth_status() |
| 380 | assert status == grok_x.AUTH_EXPIRED |
| 381 | assert "expired" in detail.lower() |
| 382 | assert expires_at is not None |
| 383 | |
| 384 | |
| 385 | def test_stored_auth_status_no_expires_at_is_ok(monkeypatch, tmp_path): |
| 386 | """Credentials without expires_at default to AUTH_OK (legacy stores).""" |
| 387 | store = tmp_path / "auth.json" |
| 388 | store.write_text('{"iss": {"refresh_token": "tok"}}') |
| 389 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 390 | status, _, expires_at = grok_x.stored_auth_status() |
| 391 | assert status == grok_x.AUTH_OK |
| 392 | assert expires_at is None |
| 393 | |
| 394 | |
| 395 | def test_stored_auth_status_unparseable_json_is_ok_with_markers(monkeypatch, tmp_path): |
| 396 | """Malformed JSON with credential markers reports AUTH_OK (graceful degradation).""" |
| 397 | store = tmp_path / "auth.json" |
| 398 | store.write_text('{"refresh_token": "tok" this is not valid json') |
| 399 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 400 | status, _, _ = grok_x.stored_auth_status() |
| 401 | assert status == grok_x.AUTH_OK |
| 402 | |
| 403 | |
| 404 | def test_stored_auth_status_unparseable_expires_at_is_ok(monkeypatch, tmp_path): |
| 405 | """Malformed expires_at is ignored, status is AUTH_OK.""" |
| 406 | store = tmp_path / "auth.json" |
| 407 | store.write_text('{"iss": {"refresh_token": "tok", "expires_at": "not-a-date"}}') |
| 408 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 409 | status, _, expires_at = grok_x.stored_auth_status() |
| 410 | assert status == grok_x.AUTH_OK |
| 411 | assert expires_at is None |
| 412 | |
| 413 | |
| 414 | def test_stored_auth_status_z_suffix_parses_correctly(monkeypatch, tmp_path): |
| 415 | """ISO 8601 timestamps with Z suffix parse correctly.""" |
| 416 | from datetime import datetime, timezone, timedelta |
| 417 | past = (datetime.now(timezone.utc) - timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ") |
| 418 | store = tmp_path / "auth.json" |
| 419 | store.write_text(f'{{"iss": {{"refresh_token": "tok", "expires_at": "{past}"}}}}') |
| 420 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 421 | status, _, _ = grok_x.stored_auth_status() |
| 422 | assert status == grok_x.AUTH_EXPIRED |
| 423 | |
| 424 | |
| 425 | def test_has_stored_auth_true_when_expired(monkeypatch, tmp_path): |
| 426 | """has_stored_auth returns True even when expired (refresh may work).""" |
| 427 | from datetime import datetime, timezone, timedelta |
| 428 | past = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat() |
| 429 | store = tmp_path / "auth.json" |
| 430 | store.write_text(f'{{"iss": {{"refresh_token": "tok", "expires_at": "{past}"}}}}') |
| 431 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 432 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 433 | assert grok_x.has_stored_auth() is True |
| 434 | |
| 435 | |
| 436 | def test_is_available_true_when_expired(monkeypatch, tmp_path): |
| 437 | """is_available returns True when expired (CLI will try refresh at runtime).""" |
| 438 | from datetime import datetime, timezone, timedelta |
| 439 | past = (datetime.now(timezone.utc) - timedelta(hours=2)).isoformat() |
| 440 | store = tmp_path / "auth.json" |
| 441 | store.write_text(f'{{"iss": {{"refresh_token": "tok", "expires_at": "{past}"}}}}') |
| 442 | monkeypatch.setattr(grok_x, "token_store_path", lambda: store) |
| 443 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 444 | grok_x.clear_availability_cache() |
| 445 | assert grok_x.is_available() is True |
| 446 | |
| 447 | |
| 448 | # --- auth revocation detection ---------------------------------------------- |
| 449 | |
| 450 | def test_is_auth_revoked_error_detects_markers(): |
| 451 | """Auth revocation markers are detected.""" |
| 452 | assert grok_x.is_auth_revoked_error("Not signed in") |
| 453 | assert grok_x.is_auth_revoked_error("invalid_grant: Refresh token has been revoked") |
| 454 | assert grok_x.is_auth_revoked_error("Authentication failed") |
| 455 | assert grok_x.is_auth_revoked_error("grok CLI exited 1: not logged in") |
| 456 | assert not grok_x.is_auth_revoked_error("timed out after 30s") |
| 457 | assert not grok_x.is_auth_revoked_error("") |
| 458 | |
| 459 | |
| 460 | def test_classify_run_failure_returns_auth_failed_for_revocation(): |
| 461 | """classify_run_failure maps revocation errors to AUTH_FAILED.""" |
| 462 | from lib import health |
| 463 | assert grok_x.classify_run_failure("Not signed in") == health.AUTH_FAILED |
| 464 | assert grok_x.classify_run_failure("invalid_grant") == health.AUTH_FAILED |
| 465 | assert grok_x.classify_run_failure("timed out") == health.TIMEOUT |
| 466 | assert grok_x.classify_run_failure("some other error") == health.ERROR |
| 467 | |
| 468 | |
| 469 | def test_search_x_returns_auth_revoked_on_session_failure(monkeypatch): |
| 470 | """search_x returns auth_revoked when the session is revoked mid-run.""" |
| 471 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 472 | monkeypatch.setattr( |
| 473 | grok_x.subprocess, "run", |
| 474 | lambda cmd, **kw: subprocess.CompletedProcess(cmd, 1, "", "Not signed in"), |
| 475 | ) |
| 476 | result = grok_x.search_x("test topic", *WINDOW) |
| 477 | assert result.get("auth_revoked") is True |
| 478 | assert "error" in result |
| 479 | |
| 480 | |
| 481 | def test_availability_cache_is_resettable(monkeypatch): |
| 482 | monkeypatch.setattr(grok_x, "_is_available_uncached", lambda: True) |
| 483 | assert grok_x.is_available() is True |
| 484 | monkeypatch.setattr(grok_x, "_is_available_uncached", lambda: False) |
| 485 | assert grok_x.is_available() is True, "memoized within a process" |
| 486 | grok_x.clear_availability_cache() |
| 487 | assert grok_x.is_available() is False |
| 488 | |
| 489 | |
| 490 | # --- lanes ----------------------------------------------------------------- |
| 491 | |
| 492 | def _stub_response(monkeypatch, body): |
| 493 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 494 | monkeypatch.setattr( |
| 495 | grok_x.subprocess, "run", |
| 496 | lambda cmd, **kw: subprocess.CompletedProcess(cmd, 0, body, ""), |
| 497 | ) |
| 498 | |
| 499 | |
| 500 | def test_from_lane_filters_by_actual_author(monkeypatch): |
| 501 | """Operator fidelity is not guaranteed: a measured from: query returned a |
| 502 | post by a different account.""" |
| 503 | _stub_response(monkeypatch, _block("2087568620465607078", handle="leojr94_")) |
| 504 | items, revoked = grok_x.search_handles(["steipete"], "topic", *WINDOW) |
| 505 | assert items == [] |
| 506 | assert revoked is False |
| 507 | |
| 508 | |
| 509 | def test_from_lane_keeps_matching_author(monkeypatch): |
| 510 | _stub_response(monkeypatch, _block("2087568620465607078", handle="steipete")) |
| 511 | items, revoked = grok_x.search_handles(["steipete"], "topic", *WINDOW) |
| 512 | assert len(items) == 1 |
| 513 | assert revoked is False |
| 514 | |
| 515 | |
| 516 | def test_from_lane_does_not_and_the_topic_into_the_query(monkeypatch): |
| 517 | seen = {} |
| 518 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 519 | |
| 520 | def fake_run(cmd, **kwargs): |
| 521 | seen["prompt"] = cmd[2] |
| 522 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 523 | |
| 524 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 525 | grok_x.search_handles(["steipete"], "quantum widgets", *WINDOW) |
| 526 | assert "quantum widgets" not in seen["prompt"] |
| 527 | |
| 528 | |
| 529 | def test_mention_lane_excludes_the_subject_client_side(monkeypatch): |
| 530 | """A measured run carrying -from:X still returned a post authored by X.""" |
| 531 | _stub_response(monkeypatch, _block("2087568620465607078", handle="GetEnergy_")) |
| 532 | items, revoked = grok_x.search_mentions(["GetEnergy_"], *WINDOW) |
| 533 | assert items == [] |
| 534 | assert revoked is False |
| 535 | |
| 536 | |
| 537 | def test_name_lane_needs_no_handle(monkeypatch): |
| 538 | _stub_response(monkeypatch, _block("2087568620465607078", handle="iamcaroren")) |
| 539 | items, revoked = grok_x.search_name("Bentgo", *WINDOW) |
| 540 | assert len(items) == 1 |
| 541 | assert revoked is False |
| 542 | |
| 543 | |
| 544 | def test_name_lane_quotes_multi_word_names(monkeypatch): |
| 545 | seen = {} |
| 546 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 547 | |
| 548 | def fake_run(cmd, **kwargs): |
| 549 | seen["prompt"] = cmd[2] |
| 550 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 551 | |
| 552 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 553 | grok_x.search_name("Peter Steinberger", *WINDOW) |
| 554 | assert '"Peter Steinberger"' in seen["prompt"] |
| 555 | |
| 556 | |
| 557 | def test_name_lane_excludes_subject_authored_posts(monkeypatch): |
| 558 | _stub_response(monkeypatch, _block("2087568620465607078", handle="Bentgo")) |
| 559 | items, revoked = grok_x.search_name("Bentgo", *WINDOW, exclude_handles=["Bentgo"]) |
| 560 | assert items == [] |
| 561 | assert revoked is False |
| 562 | |
| 563 | |
| 564 | def test_name_lane_applies_an_engagement_floor(monkeypatch): |
| 565 | seen = {} |
| 566 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 567 | |
| 568 | def fake_run(cmd, **kwargs): |
| 569 | seen["prompt"] = cmd[2] |
| 570 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 571 | |
| 572 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 573 | grok_x.search_name("Bentgo", *WINDOW) |
| 574 | assert "min_faves:" in seen["prompt"], ( |
| 575 | "the bare-name lane is the widest of the three and needs a floor the " |
| 576 | "other two do not" |
| 577 | ) |
| 578 | |
| 579 | |
| 580 | # --- lane revocation propagation ------------------------------------------- |
| 581 | |
| 582 | def test_from_lane_returns_revoked_on_auth_failure(monkeypatch): |
| 583 | """search_handles should return (items, True) when auth is revoked mid-lane.""" |
| 584 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 585 | |
| 586 | def fake_run(cmd, **kwargs): |
| 587 | return subprocess.CompletedProcess(cmd, 1, "", "Error: Not signed in") |
| 588 | |
| 589 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 590 | items, revoked = grok_x.search_handles(["steipete"], "topic", *WINDOW) |
| 591 | assert revoked is True |
| 592 | assert items == [] |
| 593 | |
| 594 | |
| 595 | def test_mention_lane_returns_revoked_on_auth_failure(monkeypatch): |
| 596 | """search_mentions should return (items, True) when auth is revoked.""" |
| 597 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 598 | |
| 599 | def fake_run(cmd, **kwargs): |
| 600 | return subprocess.CompletedProcess(cmd, 1, "", "Error: Not signed in") |
| 601 | |
| 602 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 603 | items, revoked = grok_x.search_mentions(["steipete"], *WINDOW) |
| 604 | assert revoked is True |
| 605 | assert items == [] |
| 606 | |
| 607 | |
| 608 | def test_name_lane_returns_revoked_on_auth_failure(monkeypatch): |
| 609 | """search_name should return (items, True) when auth is revoked.""" |
| 610 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 611 | |
| 612 | def fake_run(cmd, **kwargs): |
| 613 | return subprocess.CompletedProcess(cmd, 1, "", "Error: Not signed in") |
| 614 | |
| 615 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 616 | items, revoked = grok_x.search_name("Bentgo", *WINDOW) |
| 617 | assert revoked is True |
| 618 | assert items == [] |
| 619 | |
| 620 | |
| 621 | def test_from_lane_preserves_items_collected_before_revocation(monkeypatch): |
| 622 | """Items collected before auth revocation should be returned with revoked=True.""" |
| 623 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 624 | call_count = {"n": 0} |
| 625 | |
| 626 | def fake_run(cmd, **kwargs): |
| 627 | call_count["n"] += 1 |
| 628 | if call_count["n"] == 1: |
| 629 | # First handle succeeds |
| 630 | return subprocess.CompletedProcess( |
| 631 | cmd, 0, _block("2087568620465607078", handle="steipete"), "" |
| 632 | ) |
| 633 | else: |
| 634 | # Second handle hits auth revocation |
| 635 | return subprocess.CompletedProcess(cmd, 1, "", "Error: Not signed in") |
| 636 | |
| 637 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 638 | items, revoked = grok_x.search_handles(["steipete", "other"], "topic", *WINDOW) |
| 639 | # Should return items from first successful call AND signal revocation |
| 640 | assert len(items) == 1 |
| 641 | assert items[0]["author_handle"] == "steipete" |
| 642 | assert revoked is True |
| 643 | |
| 644 | |
| 645 | # --- fixes applied after review -------------------------------------------- |
| 646 | |
| 647 | def test_child_home_is_not_the_users_home(monkeypatch, tmp_path): |
| 648 | """Stripping credential env vars is not enough: the engine writes those same |
| 649 | credentials to $HOME/.config/last30days/.env, and an empty cwd is no |
| 650 | boundary for a filesystem-capable child (cwd bounds relative paths, not |
| 651 | $HOME/... reads).""" |
| 652 | seen = {} |
| 653 | |
| 654 | def fake_run(cmd, **kwargs): |
| 655 | seen["env"] = kwargs.get("env") |
| 656 | seen["cwd"] = kwargs.get("cwd") |
| 657 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 658 | |
| 659 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 660 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 661 | grok_x.search_x("steipete", *WINDOW) |
| 662 | import os as _os |
| 663 | assert seen["env"]["HOME"] != _os.path.expanduser("~") |
| 664 | assert seen["env"]["HOME"].startswith(seen["cwd"]) |
| 665 | |
| 666 | |
| 667 | def test_abbreviated_engagement_does_not_invert_ranking(): |
| 668 | """'1.2M' parsed as 1 ranked a viral post below one with 500 likes.""" |
| 669 | assert grok_x._as_int("39K") == 39_000 |
| 670 | assert grok_x._as_int("1.2M") == 1_200_000 |
| 671 | assert grok_x._as_int("1,462") == 1462 |
| 672 | assert grok_x._as_int("N/A") is None |
| 673 | |
| 674 | |
| 675 | def test_unparsable_window_rejects_rather_than_bypasses(): |
| 676 | """Fail closed: a bad window must not silently disable provenance.""" |
| 677 | assert grok_x.parse_x_response( |
| 678 | {"text": _block("2087568620465607078")}, "steipete", "not-a-date", "also-bad" |
| 679 | ) == [] |
| 680 | |
| 681 | |
| 682 | def test_handle_outside_x_grammar_is_rejected(): |
| 683 | """Model-reported handles reach post URLs and the next child's prompt.""" |
| 684 | text = _block("2087568620465607078", handle="Peter Steinberger (@steipete)") |
| 685 | items = grok_x.parse_x_response({"text": text}, "steipete", *WINDOW) |
| 686 | # Falls back to the @-pattern inside the value, or drops the item entirely. |
| 687 | assert all( |
| 688 | re.fullmatch(r"[A-Za-z0-9_]{1,15}", i["author_handle"]) for i in items |
| 689 | ) |
| 690 | |
| 691 | |
| 692 | def test_clean_handle_rejects_non_grammar_values(): |
| 693 | assert grok_x._clean_handle("@steipete") == "steipete" |
| 694 | assert grok_x._clean_handle("Peter Steinberger") == "" |
| 695 | assert grok_x._clean_handle("a" * 16) == "" |
| 696 | assert grok_x._clean_handle("bad'; drop") == "" |
| 697 | |
| 698 | |
| 699 | def test_empty_result_is_not_reported_as_an_error(): |
| 700 | """A quiet window is not a broken backend.""" |
| 701 | import subprocess as sp |
| 702 | import unittest.mock as m |
| 703 | with m.patch.object(grok_x, "binary_path", lambda: "/usr/bin/grok"), \ |
| 704 | m.patch.object(grok_x.subprocess, "run", |
| 705 | lambda cmd, **kw: sp.CompletedProcess(cmd, 0, "no posts found", "")): |
| 706 | result = grok_x.search_x("nothing-matches-this", *WINDOW) |
| 707 | assert result["items"] == [] |
| 708 | assert "error" not in result |
| 709 | |
| 710 | |
| 711 | def test_depth_drives_the_fanout_call_count(): |
| 712 | """DEPTH_CONFIG was dead: grok returned 10 posts at every depth while |
| 713 | sitting ahead of bird, silently downgrading a deep run.""" |
| 714 | quick = grok_x._fanout_queries("t", "2026-07-14", "2026-08-13", 1) |
| 715 | deep = grok_x._fanout_queries("t", "2026-07-14", "2026-08-13", 4) |
| 716 | assert len(quick) == 1 and len(deep) == 4 |
| 717 | assert len(set(deep)) == 4, "fan-out variants must differ or they repeat one result set" |
| 718 | |
| 719 | |
| 720 | def test_fanout_queries_no_phrase_quote_for_place_names(): |
| 721 | """_fanout_queries("Rome Italy") must not emit '"Rome Italy"' variant. |
| 722 | |
| 723 | This was the 2026-08-14 Rome failure: phrase-quoting "Rome Italy" returned |
| 724 | thin hits that promoted off-topic accounts (PrettyCitiesX, visegrad24). |
| 725 | """ |
| 726 | variants = grok_x._fanout_queries("Rome Italy", "2026-07-14", "2026-08-13", 4) |
| 727 | for v in variants: |
| 728 | assert '"Rome Italy"' not in v, ( |
| 729 | "Place/disambiguation strings must not be phrase-quoted; " |
| 730 | f"got {v!r}" |
| 731 | ) |
| 732 | # First variant should use unquoted AND |
| 733 | assert "Rome Italy" in variants[0] |
| 734 | |
| 735 | |
| 736 | def test_fanout_queries_proper_name_gets_phrase_quote(): |
| 737 | """Proper names like 'Peter Steinberger' should get phrase-quoted variant.""" |
| 738 | variants = grok_x._fanout_queries("Peter Steinberger", "2026-07-14", "2026-08-13", 4) |
| 739 | # One of the variants should phrase-quote the proper name |
| 740 | has_phrase = any('"Peter Steinberger"' in v for v in variants) |
| 741 | assert has_phrase, ( |
| 742 | "Proper person names should have a phrase-quoted variant for exact match" |
| 743 | ) |
| 744 | |
| 745 | |
| 746 | def test_search_handles_and_topic_false_does_not_add_topic(monkeypatch): |
| 747 | """search_handles without and_topic should not AND the topic into query.""" |
| 748 | seen = {} |
| 749 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 750 | |
| 751 | def fake_run(cmd, **kwargs): |
| 752 | seen["prompt"] = cmd[2] |
| 753 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 754 | |
| 755 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 756 | grok_x.search_handles(["steipete"], "quantum widgets", *WINDOW, and_topic=False) |
| 757 | assert "quantum widgets" not in seen["prompt"] |
| 758 | |
| 759 | |
| 760 | def test_search_handles_and_topic_true_adds_topic_to_query(monkeypatch): |
| 761 | """search_handles with and_topic=True should AND the topic into query.""" |
| 762 | seen = {} |
| 763 | monkeypatch.setattr(grok_x, "binary_path", lambda: "/usr/bin/grok") |
| 764 | |
| 765 | def fake_run(cmd, **kwargs): |
| 766 | seen["prompt"] = cmd[2] |
| 767 | return subprocess.CompletedProcess(cmd, 0, _block("2087568620465607078"), "") |
| 768 | |
| 769 | monkeypatch.setattr(grok_x.subprocess, "run", fake_run) |
| 770 | grok_x.search_handles(["visegrad24"], "Rome", *WINDOW, and_topic=True) |
| 771 | assert "Rome" in seen["prompt"], ( |
| 772 | "Extracted handles should AND the topic to ensure on-topic results" |
| 773 | ) |
| 774 |