| 1 | """Tests for xurl_x module.""" |
| 2 | |
| 3 | import json |
| 4 | import tempfile |
| 5 | import unittest |
| 6 | from pathlib import Path |
| 7 | from unittest import mock |
| 8 | |
| 9 | from lib import xurl_x |
| 10 | |
| 11 | # --------------------------------------------------------------------------- |
| 12 | # Helpers |
| 13 | # --------------------------------------------------------------------------- |
| 14 | |
| 15 | |
| 16 | def _make_api_response(tweets=None, users=None): |
| 17 | """Build a minimal X API v2 search/recent response.""" |
| 18 | tweets = tweets or [] |
| 19 | users = users or [] |
| 20 | resp = {"data": tweets} |
| 21 | if users: |
| 22 | resp["includes"] = {"users": users} |
| 23 | return resp |
| 24 | |
| 25 | # --------------------------------------------------------------------------- |
| 26 | # is_available |
| 27 | # --------------------------------------------------------------------------- |
| 28 | |
| 29 | |
| 30 | class TestIsAvailable(unittest.TestCase): |
| 31 | def setUp(self): |
| 32 | # is_available() memoizes per process; isolate every test. |
| 33 | xurl_x.clear_availability_cache() |
| 34 | self.addCleanup(xurl_x.clear_availability_cache) |
| 35 | |
| 36 | def test_returns_true_when_bearer_configured(self): |
| 37 | completed = mock.Mock( |
| 38 | returncode=0, |
| 39 | stdout="oauth1: ✗\nbearer: ✓\n", |
| 40 | ) |
| 41 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 42 | self.assertTrue(xurl_x.is_available()) |
| 43 | call_args = run_mock.call_args[0][0] |
| 44 | self.assertEqual(call_args[:3], ["xurl", "auth", "status"]) |
| 45 | |
| 46 | def test_returns_false_when_oauth1_only(self): |
| 47 | # OAuth1 alone cannot satisfy search_x's --auth app requirement. |
| 48 | completed = mock.Mock( |
| 49 | returncode=0, |
| 50 | stdout="oauth1: ✓\nbearer: ✗\n", |
| 51 | ) |
| 52 | with mock.patch("subprocess.run", return_value=completed): |
| 53 | self.assertFalse(xurl_x.is_available()) |
| 54 | |
| 55 | def test_returns_false_when_not_authenticated(self): |
| 56 | completed = mock.Mock(returncode=1, stdout="") |
| 57 | with mock.patch("subprocess.run", return_value=completed): |
| 58 | self.assertFalse(xurl_x.is_available()) |
| 59 | |
| 60 | def test_returns_false_when_not_installed(self): |
| 61 | with mock.patch("subprocess.run", side_effect=FileNotFoundError): |
| 62 | self.assertFalse(xurl_x.is_available()) |
| 63 | |
| 64 | def test_returns_false_on_permission_error(self): |
| 65 | # WSL hits this when a Windows-mounted PATH entry points at an |
| 66 | # exec-blocked shim (e.g. WindowsApps), which raises PermissionError |
| 67 | # before any other PATH candidate is tried. |
| 68 | with mock.patch("subprocess.run", side_effect=PermissionError(13, "Permission denied", "xurl")): |
| 69 | self.assertFalse(xurl_x.is_available()) |
| 70 | |
| 71 | def test_returns_false_on_timeout(self): |
| 72 | import subprocess |
| 73 | with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("xurl", 10)): |
| 74 | self.assertFalse(xurl_x.is_available()) |
| 75 | |
| 76 | def test_returns_false_when_no_bearer_marker(self): |
| 77 | # returncode=0 but status output has no bearer: ✓ |
| 78 | completed = mock.Mock(returncode=0, stdout="oauth1: ✗\nbearer: ✗\n") |
| 79 | with mock.patch("subprocess.run", return_value=completed): |
| 80 | self.assertFalse(xurl_x.is_available()) |
| 81 | |
| 82 | # --------------------------------------------------------------------------- |
| 83 | # stored_auth_status / has_stored_auth (local-only doctor-path evidence) |
| 84 | # --------------------------------------------------------------------------- |
| 85 | |
| 86 | |
| 87 | class _UnreadableStore: |
| 88 | """Path stub: exists but every read raises (permission-denied store).""" |
| 89 | |
| 90 | def is_file(self): |
| 91 | return True |
| 92 | |
| 93 | def read_text(self, *args, **kwargs): |
| 94 | raise PermissionError(13, "Permission denied") |
| 95 | |
| 96 | def __str__(self): |
| 97 | return "/home/user/.xurl" |
| 98 | |
| 99 | |
| 100 | class TestStoredAuth(unittest.TestCase): |
| 101 | """F1/F10: the doctor path keys on xurl's on-disk token store (~/.xurl) |
| 102 | instead of the live `xurl whoami` network call. These tests forbid |
| 103 | subprocess entirely — local evidence must never spawn anything.""" |
| 104 | |
| 105 | def setUp(self): |
| 106 | self._tmp = tempfile.TemporaryDirectory() |
| 107 | self.addCleanup(self._tmp.cleanup) |
| 108 | self.store = Path(self._tmp.name) / ".xurl" |
| 109 | boom = mock.patch( |
| 110 | "subprocess.run", |
| 111 | side_effect=AssertionError("local auth evidence must not spawn a subprocess"), |
| 112 | ) |
| 113 | boom.start() |
| 114 | self.addCleanup(boom.stop) |
| 115 | |
| 116 | def _status(self, path=None): |
| 117 | with mock.patch("lib.xurl_x.token_store_path", return_value=path or self.store): |
| 118 | return xurl_x.stored_auth_status() |
| 119 | |
| 120 | def test_yaml_store_with_access_token_is_ok(self): |
| 121 | self.store.write_text( |
| 122 | "apps:\n app:\n oauth2_tokens:\n me:\n oauth2:\n" |
| 123 | " access_token: dummy-not-real\n", |
| 124 | encoding="utf-8", |
| 125 | ) |
| 126 | status, detail = self._status() |
| 127 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 128 | self.assertIn(str(self.store), detail) |
| 129 | |
| 130 | def test_directory_layout_auth_yml_with_token_is_ok(self): |
| 131 | # Current xurl (>=1.1) stores credentials at ~/.xurl/auth.yml inside |
| 132 | # the ~/.xurl directory. Regression for #978: this was misread as |
| 133 | # "no token store" because the directory itself fails is_file(). |
| 134 | self.store.mkdir(exist_ok=True) |
| 135 | auth_yml = self.store / "auth.yml" |
| 136 | auth_yml.write_text( |
| 137 | "apps:\n app:\n oauth2_tokens:\n me:\n oauth2:\n" |
| 138 | " access_token: dummy-not-real\n", |
| 139 | encoding="utf-8", |
| 140 | ) |
| 141 | status, detail = self._status() |
| 142 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 143 | self.assertIn("auth.yml", detail) |
| 144 | |
| 145 | def test_directory_layout_empty_auth_yml_is_missing(self): |
| 146 | self.store.mkdir(exist_ok=True) |
| 147 | (self.store / "auth.yml").write_text("", encoding="utf-8") |
| 148 | status, _ = self._status() |
| 149 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 150 | |
| 151 | def test_directory_layout_without_auth_yml_is_missing(self): |
| 152 | self.store.mkdir(exist_ok=True) |
| 153 | status, detail = self._status() |
| 154 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 155 | self.assertIn("no token store", detail) |
| 156 | |
| 157 | def test_legacy_flat_file_layout_is_ok(self): |
| 158 | # When token_store_path() returns the canonical ~/.xurl/auth.yml but |
| 159 | # the legacy flat ~/.xurl file is what exists, still report OK. |
| 160 | self.store.write_text( |
| 161 | json.dumps({"bearer_token": {"bearer": "dummy-not-real"}}), |
| 162 | encoding="utf-8", |
| 163 | ) |
| 164 | status, detail = self._status() |
| 165 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 166 | self.assertIn(str(self.store), detail) |
| 167 | |
| 168 | def test_directory_layout_unreadable_auth_yml_is_error(self): |
| 169 | self.store.mkdir(exist_ok=True) |
| 170 | (self.store / "auth.yml").write_text( |
| 171 | "access_token: dummy-not-real\n", encoding="utf-8" |
| 172 | ) |
| 173 | class _BrokenFile: |
| 174 | def __init__(self, path): |
| 175 | self._path = path |
| 176 | |
| 177 | def is_file(self): |
| 178 | return True |
| 179 | |
| 180 | def read_text(self, *args, **kwargs): |
| 181 | raise PermissionError(13, "Permission denied") |
| 182 | |
| 183 | def __str__(self): |
| 184 | return str(self._path) |
| 185 | |
| 186 | status, detail = self._status( |
| 187 | path=_BrokenFile(self.store / "auth.yml") |
| 188 | ) |
| 189 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 190 | self.assertIn("PermissionError", detail) |
| 191 | |
| 192 | def test_permission_denied_read_reports_error_not_missing(self): |
| 193 | # Regression: a store that exists but cannot be read must report the |
| 194 | # typed AUTH_ERROR, not AUTH_MISSING. stat() on a chmod-000 file still |
| 195 | # succeeds on POSIX; read_text() is what raises PermissionError. |
| 196 | import os |
| 197 | |
| 198 | if hasattr(os, "geteuid") and os.geteuid() == 0: |
| 199 | self.skipTest("root bypasses permission checks") |
| 200 | self.store.mkdir(exist_ok=True) |
| 201 | auth_yml = self.store / "auth.yml" |
| 202 | auth_yml.write_text("access_token: dummy-not-real\n", encoding="utf-8") |
| 203 | auth_yml.chmod(0) |
| 204 | try: |
| 205 | status, detail = self._status() |
| 206 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 207 | self.assertIn("PermissionError", detail) |
| 208 | finally: |
| 209 | auth_yml.chmod(0o600) |
| 210 | |
| 211 | def test_permission_denied_parent_stat_reports_error_not_missing(self): |
| 212 | # Regression: a store inside a non-searchable directory must report |
| 213 | # AUTH_ERROR, not AUTH_MISSING. stat() on the candidate inside a |
| 214 | # chmod-000 parent raises PermissionError, which must not be swallowed |
| 215 | # into "no token store" the way pathlib's is_file() would. |
| 216 | import os |
| 217 | |
| 218 | if hasattr(os, "geteuid") and os.geteuid() == 0: |
| 219 | self.skipTest("root bypasses permission checks") |
| 220 | self.store.mkdir(exist_ok=True) |
| 221 | (self.store / "auth.yml").write_text( |
| 222 | "access_token: dummy-not-real\n", encoding="utf-8" |
| 223 | ) |
| 224 | self.store.chmod(0) |
| 225 | try: |
| 226 | status, detail = self._status() |
| 227 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 228 | self.assertIn("PermissionError", detail) |
| 229 | finally: |
| 230 | self.store.chmod(0o700) |
| 231 | |
| 232 | def test_permission_denied_grandparent_stat_reports_error_not_missing(self): |
| 233 | # Regression: distinct from the parent-chmod case above, which denies |
| 234 | # traversal INTO the store (raising on the per-candidate _is_file scan). |
| 235 | # Chmod-000 on the store's own PARENT directory instead blocks stat() |
| 236 | # on the store path itself, raising during the _is_dir(base) call that |
| 237 | # builds the candidate list -- a separate except-OSError branch that |
| 238 | # the parent-chmod case never reaches. |
| 239 | import os |
| 240 | |
| 241 | if hasattr(os, "geteuid") and os.geteuid() == 0: |
| 242 | self.skipTest("root bypasses permission checks") |
| 243 | self.store.mkdir(exist_ok=True) |
| 244 | (self.store / "auth.yml").write_text( |
| 245 | "access_token: dummy-not-real\n", encoding="utf-8" |
| 246 | ) |
| 247 | grandparent = self.store.parent |
| 248 | grandparent.chmod(0) |
| 249 | try: |
| 250 | status, detail = self._status() |
| 251 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 252 | self.assertIn("PermissionError", detail) |
| 253 | finally: |
| 254 | grandparent.chmod(0o700) |
| 255 | |
| 256 | def test_directory_layout_has_stored_auth_with_binary(self): |
| 257 | self.store.mkdir(exist_ok=True) |
| 258 | (self.store / "auth.yml").write_text( |
| 259 | "access_token: dummy-not-real\n", encoding="utf-8" |
| 260 | ) |
| 261 | with mock.patch("lib.xurl_x.token_store_path", return_value=self.store), \ |
| 262 | mock.patch("lib.xurl_x.shutil.which", return_value="/usr/local/bin/xurl"): |
| 263 | self.assertTrue(xurl_x.has_stored_auth()) |
| 264 | |
| 265 | def test_directory_layout_no_subprocess_spawned(self): |
| 266 | self.store.mkdir(exist_ok=True) |
| 267 | (self.store / "auth.yml").write_text( |
| 268 | "access_token: dummy-not-real\n", encoding="utf-8" |
| 269 | ) |
| 270 | with mock.patch( |
| 271 | "subprocess.run", |
| 272 | side_effect=AssertionError("local auth evidence must not spawn a subprocess"), |
| 273 | ): |
| 274 | status, _ = self._status() |
| 275 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 276 | |
| 277 | def test_walk_up_canonical_path_finds_legacy_flat_file(self): |
| 278 | # token_store_path() returns the canonical ~/.xurl/auth.yml, but only |
| 279 | # the legacy flat ~/.xurl file exists. The parent-walk must find it. |
| 280 | self.store.write_text( |
| 281 | json.dumps({"bearer_token": {"bearer": "dummy-not-real"}}), |
| 282 | encoding="utf-8", |
| 283 | ) |
| 284 | canonical = self.store / "auth.yml" |
| 285 | with mock.patch("lib.xurl_x.token_store_path", return_value=canonical): |
| 286 | status, detail = xurl_x.stored_auth_status() |
| 287 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 288 | self.assertIn(str(self.store), detail) |
| 289 | |
| 290 | def test_canonical_path_preferred_when_both_layouts_exist(self): |
| 291 | # A stale legacy flat ~/.xurl must never shadow the live auth.yml. |
| 292 | self.store.mkdir(exist_ok=True) |
| 293 | (self.store / "auth.yml").write_text( |
| 294 | "oauth2_tokens:\n me:\n oauth2:\n access_token: live\n", |
| 295 | encoding="utf-8", |
| 296 | ) |
| 297 | canonical = self.store / "auth.yml" |
| 298 | with mock.patch("lib.xurl_x.token_store_path", return_value=canonical): |
| 299 | status, detail = xurl_x.stored_auth_status() |
| 300 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 301 | self.assertIn("auth.yml", detail) |
| 302 | |
| 303 | def test_legacy_json_store_with_bearer_token_is_ok(self): |
| 304 | self.store.write_text( |
| 305 | json.dumps({"bearer_token": {"bearer": "dummy-not-real"}}), |
| 306 | encoding="utf-8", |
| 307 | ) |
| 308 | status, _ = self._status() |
| 309 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 310 | |
| 311 | def test_absent_store_is_missing(self): |
| 312 | status, detail = self._status() |
| 313 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 314 | self.assertIn("no token store", detail) |
| 315 | |
| 316 | def test_empty_store_is_missing(self): |
| 317 | self.store.write_text("", encoding="utf-8") |
| 318 | status, _ = self._status() |
| 319 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 320 | |
| 321 | def test_store_without_credential_markers_is_missing(self): |
| 322 | self.store.write_text("apps: {}\ndefault_app: app\n", encoding="utf-8") |
| 323 | status, detail = self._status() |
| 324 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 325 | self.assertIn("no stored credentials", detail) |
| 326 | |
| 327 | def test_unreadable_store_is_error_not_missing(self): |
| 328 | status, detail = self._status(path=_UnreadableStore()) |
| 329 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 330 | self.assertIn("unreadable", detail) |
| 331 | self.assertIn("PermissionError", detail) |
| 332 | |
| 333 | def test_has_stored_auth_true_with_binary_and_store(self): |
| 334 | self.store.write_text("access_token: dummy-not-real\n", encoding="utf-8") |
| 335 | with mock.patch("lib.xurl_x.token_store_path", return_value=self.store), \ |
| 336 | mock.patch("lib.xurl_x.shutil.which", return_value="/usr/local/bin/xurl"): |
| 337 | self.assertTrue(xurl_x.has_stored_auth()) |
| 338 | |
| 339 | def test_has_stored_auth_false_without_binary(self): |
| 340 | self.store.write_text("access_token: dummy-not-real\n", encoding="utf-8") |
| 341 | with mock.patch("lib.xurl_x.token_store_path", return_value=self.store), \ |
| 342 | mock.patch("lib.xurl_x.shutil.which", return_value=None): |
| 343 | self.assertFalse(xurl_x.has_stored_auth()) |
| 344 | |
| 345 | def test_has_stored_auth_false_on_broken_store(self): |
| 346 | # has_stored_auth answers availability only; the typed ERROR surface |
| 347 | # lives in backends._probe_xurl (see test_backend_descriptors). |
| 348 | with mock.patch("lib.xurl_x.token_store_path", return_value=_UnreadableStore()), \ |
| 349 | mock.patch("lib.xurl_x.shutil.which", return_value="/usr/local/bin/xurl"): |
| 350 | self.assertFalse(xurl_x.has_stored_auth()) |
| 351 | |
| 352 | def test_default_store_path_is_home_dot_xurl_auth_yml(self): |
| 353 | self.assertEqual(Path.home() / ".xurl" / "auth.yml", xurl_x.token_store_path()) |
| 354 | |
| 355 | # --------------------------------------------------------------------------- |
| 356 | # search_x |
| 357 | # --------------------------------------------------------------------------- |
| 358 | |
| 359 | |
| 360 | class TestSearchX(unittest.TestCase): |
| 361 | def test_returns_parsed_json_on_success(self): |
| 362 | payload = {"data": [{"id": "1", "text": "hello world", "author_id": "u1"}]} |
| 363 | completed = mock.Mock(returncode=0, stdout=json.dumps(payload)) |
| 364 | with mock.patch("subprocess.run", return_value=completed): |
| 365 | result = xurl_x.search_x("hello world") |
| 366 | self.assertEqual(result["data"][0]["id"], "1") |
| 367 | |
| 368 | def test_returns_fixed_error_on_non_zero_exit(self): |
| 369 | # xurl's stderr may echo the request or the bearer; only an |
| 370 | # engine-authored fixed string (with a classifier marker) survives. |
| 371 | completed = mock.Mock( |
| 372 | returncode=1, stdout="", |
| 373 | stderr="rate limit exceeded for dummy-x-bearer-secret-000", |
| 374 | ) |
| 375 | with mock.patch("subprocess.run", return_value=completed): |
| 376 | result = xurl_x.search_x("test") |
| 377 | self.assertEqual(xurl_x.ERR_RATE_LIMITED, result["error"]) |
| 378 | self.assertIn("rate limit", result["error"]) |
| 379 | self.assertNotIn("dummy-x-bearer-secret-000", result["error"]) |
| 380 | |
| 381 | def test_returns_fixed_error_on_invalid_json(self): |
| 382 | completed = mock.Mock(returncode=0, stdout="NOT JSON dummy-x-bearer-secret-000") |
| 383 | with mock.patch("subprocess.run", return_value=completed): |
| 384 | result = xurl_x.search_x("test") |
| 385 | self.assertEqual(xurl_x.ERR_INVALID_JSON, result["error"]) |
| 386 | self.assertIn("invalid JSON", result["error"]) |
| 387 | self.assertNotIn("dummy-x-bearer-secret-000", result["error"]) |
| 388 | |
| 389 | def test_returns_error_when_not_installed(self): |
| 390 | with mock.patch("subprocess.run", side_effect=FileNotFoundError): |
| 391 | result = xurl_x.search_x("test") |
| 392 | self.assertIn("error", result) |
| 393 | self.assertIn("not found", result["error"]) |
| 394 | |
| 395 | def test_returns_error_on_timeout(self): |
| 396 | import subprocess |
| 397 | with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("xurl", 30)): |
| 398 | result = xurl_x.search_x("test") |
| 399 | self.assertIn("error", result) |
| 400 | self.assertIn("timed out", result["error"]) |
| 401 | |
| 402 | def test_search_uses_app_only_auth(self): |
| 403 | # Regression: default (OAuth1) auth 401s on any query needing |
| 404 | # percent-encoding (xurl >=1.1 signing bug); search must pin app-only. |
| 405 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 406 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 407 | xurl_x.search_x("claude code") |
| 408 | call_args = run_mock.call_args[0][0] |
| 409 | self.assertIn("--auth", call_args) |
| 410 | self.assertEqual(call_args[call_args.index("--auth") + 1], "app") |
| 411 | |
| 412 | def test_max_results_clamped_to_100(self): |
| 413 | # DEPTH_CONFIG["deep"] = 60, should stay at 60 (within 10-100 range) |
| 414 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 415 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 416 | xurl_x.search_x("test", depth="deep") |
| 417 | call_args = run_mock.call_args[0][0] |
| 418 | n_idx = call_args.index("-n") |
| 419 | self.assertLessEqual(int(call_args[n_idx + 1]), 100) |
| 420 | |
| 421 | def test_max_results_at_least_10(self): |
| 422 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 423 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 424 | xurl_x.search_x("test", depth="quick") |
| 425 | call_args = run_mock.call_args[0][0] |
| 426 | n_idx = call_args.index("-n") |
| 427 | self.assertGreaterEqual(int(call_args[n_idx + 1]), 10) |
| 428 | |
| 429 | def test_unknown_depth_falls_back_to_default(self): |
| 430 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 431 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 432 | xurl_x.search_x("test", depth="nonexistent") |
| 433 | call_args = run_mock.call_args[0][0] |
| 434 | n_idx = call_args.index("-n") |
| 435 | self.assertEqual(int(call_args[n_idx + 1]), xurl_x.DEPTH_CONFIG["default"]) |
| 436 | |
| 437 | # --------------------------------------------------------------------------- |
| 438 | # parse_x_response |
| 439 | # --------------------------------------------------------------------------- |
| 440 | |
| 441 | |
| 442 | class TestParseXResponse(unittest.TestCase): |
| 443 | def _tweet(self, id_, text, author_id, created_at=None, metrics=None): |
| 444 | t = {"id": id_, "text": text, "author_id": author_id} |
| 445 | if created_at: |
| 446 | t["created_at"] = created_at |
| 447 | if metrics: |
| 448 | t["public_metrics"] = metrics |
| 449 | return t |
| 450 | |
| 451 | def _user(self, id_, username): |
| 452 | return {"id": id_, "username": username} |
| 453 | |
| 454 | def test_empty_response_returns_empty_list(self): |
| 455 | self.assertEqual(xurl_x.parse_x_response({}), []) |
| 456 | |
| 457 | def test_error_response_returns_empty_list(self): |
| 458 | self.assertEqual(xurl_x.parse_x_response({"error": "oops"}), []) |
| 459 | |
| 460 | def test_parses_basic_tweet(self): |
| 461 | resp = _make_api_response( |
| 462 | tweets=[self._tweet("111", "Hello AI", "u1")], |
| 463 | users=[self._user("u1", "alice")], |
| 464 | ) |
| 465 | items = xurl_x.parse_x_response(resp) |
| 466 | self.assertEqual(len(items), 1) |
| 467 | self.assertEqual(items[0]["text"], "Hello AI") |
| 468 | self.assertEqual(items[0]["author_handle"], "alice") |
| 469 | self.assertIn("alice", items[0]["url"]) |
| 470 | self.assertIn("111", items[0]["url"]) |
| 471 | |
| 472 | def test_parses_date_from_iso(self): |
| 473 | resp = _make_api_response( |
| 474 | tweets=[self._tweet("1", "text", "u1", created_at="2024-06-15T12:00:00Z")], |
| 475 | ) |
| 476 | items = xurl_x.parse_x_response(resp) |
| 477 | self.assertEqual(items[0]["date"], "2024-06-15") |
| 478 | |
| 479 | def test_date_none_when_missing(self): |
| 480 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 481 | items = xurl_x.parse_x_response(resp) |
| 482 | self.assertIsNone(items[0]["date"]) |
| 483 | |
| 484 | def test_parses_engagement_metrics(self): |
| 485 | metrics = { |
| 486 | "like_count": 42, |
| 487 | "retweet_count": 10, |
| 488 | "reply_count": 5, |
| 489 | "quote_count": 2, |
| 490 | } |
| 491 | resp = _make_api_response( |
| 492 | tweets=[self._tweet("1", "text", "u1", metrics=metrics)], |
| 493 | ) |
| 494 | items = xurl_x.parse_x_response(resp) |
| 495 | self.assertEqual(items[0]["engagement"]["likes"], 42) |
| 496 | self.assertEqual(items[0]["engagement"]["reposts"], 10) |
| 497 | self.assertEqual(items[0]["engagement"]["replies"], 5) |
| 498 | self.assertEqual(items[0]["engagement"]["quotes"], 2) |
| 499 | |
| 500 | def test_engagement_none_when_no_metrics(self): |
| 501 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 502 | items = xurl_x.parse_x_response(resp) |
| 503 | self.assertIsNone(items[0]["engagement"]) |
| 504 | |
| 505 | def test_text_truncated_to_500_chars(self): |
| 506 | long_text = "x" * 600 |
| 507 | resp = _make_api_response(tweets=[self._tweet("1", long_text, "u1")]) |
| 508 | items = xurl_x.parse_x_response(resp) |
| 509 | self.assertLessEqual(len(items[0]["text"]), 500) |
| 510 | |
| 511 | def test_id_prefixed_with_xurl(self): |
| 512 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 513 | items = xurl_x.parse_x_response(resp) |
| 514 | self.assertTrue(items[0]["id"].startswith("XURL")) |
| 515 | |
| 516 | def test_relevance_computed_when_topic_given(self): |
| 517 | resp = _make_api_response( |
| 518 | tweets=[self._tweet("1", "Claude Code is great for AI coding", "u1")], |
| 519 | ) |
| 520 | items = xurl_x.parse_x_response(resp, topic="Claude Code") |
| 521 | self.assertGreater(items[0]["relevance"], 0.5) |
| 522 | |
| 523 | def test_relevance_neutral_when_no_topic(self): |
| 524 | resp = _make_api_response(tweets=[self._tweet("1", "some text", "u1")]) |
| 525 | items = xurl_x.parse_x_response(resp) |
| 526 | self.assertEqual(items[0]["relevance"], 0.5) |
| 527 | |
| 528 | def test_url_falls_back_to_i_status_when_no_username(self): |
| 529 | # author_id not in includes.users → username="" but the post is |
| 530 | # kept with the id-only citation form (shared x_api parser). |
| 531 | resp = _make_api_response(tweets=[self._tweet("999", "text", "unknown_uid")]) |
| 532 | items = xurl_x.parse_x_response(resp) |
| 533 | self.assertEqual(items[0]["url"], "https://x.com/i/status/999") |
| 534 | self.assertEqual(items[0]["author_handle"], "") |
| 535 | self.assertEqual(items[0]["post_id"], "999") |
| 536 | |
| 537 | def test_multiple_tweets_parsed(self): |
| 538 | tweets = [self._tweet(str(i), f"tweet {i}", "u1") for i in range(5)] |
| 539 | resp = _make_api_response(tweets=tweets, users=[self._user("u1", "bob")]) |
| 540 | items = xurl_x.parse_x_response(resp) |
| 541 | self.assertEqual(len(items), 5) |
| 542 | |
| 543 | def test_empty_data_list(self): |
| 544 | resp = _make_api_response(tweets=[]) |
| 545 | self.assertEqual(xurl_x.parse_x_response(resp), []) |
| 546 | |
| 547 | def test_why_relevant_is_empty_string(self): |
| 548 | # xurl doesn't provide LLM-generated why_relevant (unlike xai_x) |
| 549 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 550 | items = xurl_x.parse_x_response(resp) |
| 551 | self.assertEqual(items[0]["why_relevant"], "") |
| 552 | |
| 553 | # --------------------------------------------------------------------------- |
| 554 | # DEPTH_CONFIG |
| 555 | # --------------------------------------------------------------------------- |
| 556 | |
| 557 | |
| 558 | class TestDepthConfig(unittest.TestCase): |
| 559 | def test_all_standard_depths_present(self): |
| 560 | for depth in ("quick", "default", "deep"): |
| 561 | self.assertIn(depth, xurl_x.DEPTH_CONFIG) |
| 562 | |
| 563 | def test_depth_config_is_shared_with_x_api(self): |
| 564 | from lib import x_api |
| 565 | self.assertIs(x_api.DEPTH_CONFIG, xurl_x.DEPTH_CONFIG) |
| 566 | |
| 567 | def test_deep_greater_than_quick(self): |
| 568 | self.assertGreater( |
| 569 | xurl_x.DEPTH_CONFIG["deep"], |
| 570 | xurl_x.DEPTH_CONFIG["quick"], |
| 571 | ) |
| 572 | |
| 573 | if __name__ == "__main__": |
| 574 | unittest.main() |
| 575 |