| 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_legacy_json_store_with_bearer_token_is_ok(self): |
| 131 | self.store.write_text( |
| 132 | json.dumps({"bearer_token": {"bearer": "dummy-not-real"}}), |
| 133 | encoding="utf-8", |
| 134 | ) |
| 135 | status, _ = self._status() |
| 136 | self.assertEqual(xurl_x.AUTH_OK, status) |
| 137 | |
| 138 | def test_absent_store_is_missing(self): |
| 139 | status, detail = self._status() |
| 140 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 141 | self.assertIn("no token store", detail) |
| 142 | |
| 143 | def test_empty_store_is_missing(self): |
| 144 | self.store.write_text("", encoding="utf-8") |
| 145 | status, _ = self._status() |
| 146 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 147 | |
| 148 | def test_store_without_credential_markers_is_missing(self): |
| 149 | self.store.write_text("apps: {}\ndefault_app: app\n", encoding="utf-8") |
| 150 | status, detail = self._status() |
| 151 | self.assertEqual(xurl_x.AUTH_MISSING, status) |
| 152 | self.assertIn("no stored credentials", detail) |
| 153 | |
| 154 | def test_unreadable_store_is_error_not_missing(self): |
| 155 | status, detail = self._status(path=_UnreadableStore()) |
| 156 | self.assertEqual(xurl_x.AUTH_ERROR, status) |
| 157 | self.assertIn("unreadable", detail) |
| 158 | self.assertIn("PermissionError", detail) |
| 159 | |
| 160 | def test_has_stored_auth_true_with_binary_and_store(self): |
| 161 | self.store.write_text("access_token: dummy-not-real\n", encoding="utf-8") |
| 162 | with mock.patch("lib.xurl_x.token_store_path", return_value=self.store), \ |
| 163 | mock.patch("lib.xurl_x.shutil.which", return_value="/usr/local/bin/xurl"): |
| 164 | self.assertTrue(xurl_x.has_stored_auth()) |
| 165 | |
| 166 | def test_has_stored_auth_false_without_binary(self): |
| 167 | self.store.write_text("access_token: dummy-not-real\n", encoding="utf-8") |
| 168 | with mock.patch("lib.xurl_x.token_store_path", return_value=self.store), \ |
| 169 | mock.patch("lib.xurl_x.shutil.which", return_value=None): |
| 170 | self.assertFalse(xurl_x.has_stored_auth()) |
| 171 | |
| 172 | def test_has_stored_auth_false_on_broken_store(self): |
| 173 | # has_stored_auth answers availability only; the typed ERROR surface |
| 174 | # lives in backends._probe_xurl (see test_backend_descriptors). |
| 175 | with mock.patch("lib.xurl_x.token_store_path", return_value=_UnreadableStore()), \ |
| 176 | mock.patch("lib.xurl_x.shutil.which", return_value="/usr/local/bin/xurl"): |
| 177 | self.assertFalse(xurl_x.has_stored_auth()) |
| 178 | |
| 179 | def test_default_store_path_is_home_dot_xurl(self): |
| 180 | self.assertEqual(Path.home() / ".xurl", xurl_x.token_store_path()) |
| 181 | |
| 182 | # --------------------------------------------------------------------------- |
| 183 | # search_x |
| 184 | # --------------------------------------------------------------------------- |
| 185 | |
| 186 | |
| 187 | class TestSearchX(unittest.TestCase): |
| 188 | def test_returns_parsed_json_on_success(self): |
| 189 | payload = {"data": [{"id": "1", "text": "hello world", "author_id": "u1"}]} |
| 190 | completed = mock.Mock(returncode=0, stdout=json.dumps(payload)) |
| 191 | with mock.patch("subprocess.run", return_value=completed): |
| 192 | result = xurl_x.search_x("hello world") |
| 193 | self.assertEqual(result["data"][0]["id"], "1") |
| 194 | |
| 195 | def test_returns_error_on_non_zero_exit(self): |
| 196 | completed = mock.Mock(returncode=1, stdout="", stderr="rate limit exceeded") |
| 197 | with mock.patch("subprocess.run", return_value=completed): |
| 198 | result = xurl_x.search_x("test") |
| 199 | self.assertIn("error", result) |
| 200 | self.assertIn("rate limit exceeded", result["error"]) |
| 201 | |
| 202 | def test_returns_error_on_invalid_json(self): |
| 203 | completed = mock.Mock(returncode=0, stdout="NOT JSON") |
| 204 | with mock.patch("subprocess.run", return_value=completed): |
| 205 | result = xurl_x.search_x("test") |
| 206 | self.assertIn("error", result) |
| 207 | self.assertIn("Invalid JSON", result["error"]) |
| 208 | |
| 209 | def test_returns_error_when_not_installed(self): |
| 210 | with mock.patch("subprocess.run", side_effect=FileNotFoundError): |
| 211 | result = xurl_x.search_x("test") |
| 212 | self.assertIn("error", result) |
| 213 | self.assertIn("not found", result["error"]) |
| 214 | |
| 215 | def test_returns_error_on_timeout(self): |
| 216 | import subprocess |
| 217 | with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired("xurl", 30)): |
| 218 | result = xurl_x.search_x("test") |
| 219 | self.assertIn("error", result) |
| 220 | self.assertIn("timed out", result["error"]) |
| 221 | |
| 222 | def test_search_uses_app_only_auth(self): |
| 223 | # Regression: default (OAuth1) auth 401s on any query needing |
| 224 | # percent-encoding (xurl >=1.1 signing bug); search must pin app-only. |
| 225 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 226 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 227 | xurl_x.search_x("claude code") |
| 228 | call_args = run_mock.call_args[0][0] |
| 229 | self.assertIn("--auth", call_args) |
| 230 | self.assertEqual(call_args[call_args.index("--auth") + 1], "app") |
| 231 | |
| 232 | def test_max_results_clamped_to_100(self): |
| 233 | # DEPTH_CONFIG["deep"] = 60, should stay at 60 (within 10-100 range) |
| 234 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 235 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 236 | xurl_x.search_x("test", depth="deep") |
| 237 | call_args = run_mock.call_args[0][0] |
| 238 | n_idx = call_args.index("-n") |
| 239 | self.assertLessEqual(int(call_args[n_idx + 1]), 100) |
| 240 | |
| 241 | def test_max_results_at_least_10(self): |
| 242 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 243 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 244 | xurl_x.search_x("test", depth="quick") |
| 245 | call_args = run_mock.call_args[0][0] |
| 246 | n_idx = call_args.index("-n") |
| 247 | self.assertGreaterEqual(int(call_args[n_idx + 1]), 10) |
| 248 | |
| 249 | def test_unknown_depth_falls_back_to_default(self): |
| 250 | completed = mock.Mock(returncode=0, stdout=json.dumps({})) |
| 251 | with mock.patch("subprocess.run", return_value=completed) as run_mock: |
| 252 | xurl_x.search_x("test", depth="nonexistent") |
| 253 | call_args = run_mock.call_args[0][0] |
| 254 | n_idx = call_args.index("-n") |
| 255 | self.assertEqual(int(call_args[n_idx + 1]), xurl_x.DEPTH_CONFIG["default"]) |
| 256 | |
| 257 | # --------------------------------------------------------------------------- |
| 258 | # parse_x_response |
| 259 | # --------------------------------------------------------------------------- |
| 260 | |
| 261 | |
| 262 | class TestParseXResponse(unittest.TestCase): |
| 263 | def _tweet(self, id_, text, author_id, created_at=None, metrics=None): |
| 264 | t = {"id": id_, "text": text, "author_id": author_id} |
| 265 | if created_at: |
| 266 | t["created_at"] = created_at |
| 267 | if metrics: |
| 268 | t["public_metrics"] = metrics |
| 269 | return t |
| 270 | |
| 271 | def _user(self, id_, username): |
| 272 | return {"id": id_, "username": username} |
| 273 | |
| 274 | def test_empty_response_returns_empty_list(self): |
| 275 | self.assertEqual(xurl_x.parse_x_response({}), []) |
| 276 | |
| 277 | def test_error_response_returns_empty_list(self): |
| 278 | self.assertEqual(xurl_x.parse_x_response({"error": "oops"}), []) |
| 279 | |
| 280 | def test_parses_basic_tweet(self): |
| 281 | resp = _make_api_response( |
| 282 | tweets=[self._tweet("111", "Hello AI", "u1")], |
| 283 | users=[self._user("u1", "alice")], |
| 284 | ) |
| 285 | items = xurl_x.parse_x_response(resp) |
| 286 | self.assertEqual(len(items), 1) |
| 287 | self.assertEqual(items[0]["text"], "Hello AI") |
| 288 | self.assertEqual(items[0]["author_handle"], "alice") |
| 289 | self.assertIn("alice", items[0]["url"]) |
| 290 | self.assertIn("111", items[0]["url"]) |
| 291 | |
| 292 | def test_parses_date_from_iso(self): |
| 293 | resp = _make_api_response( |
| 294 | tweets=[self._tweet("1", "text", "u1", created_at="2024-06-15T12:00:00Z")], |
| 295 | ) |
| 296 | items = xurl_x.parse_x_response(resp) |
| 297 | self.assertEqual(items[0]["date"], "2024-06-15") |
| 298 | |
| 299 | def test_date_none_when_missing(self): |
| 300 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 301 | items = xurl_x.parse_x_response(resp) |
| 302 | self.assertIsNone(items[0]["date"]) |
| 303 | |
| 304 | def test_parses_engagement_metrics(self): |
| 305 | metrics = { |
| 306 | "like_count": 42, |
| 307 | "retweet_count": 10, |
| 308 | "reply_count": 5, |
| 309 | "quote_count": 2, |
| 310 | } |
| 311 | resp = _make_api_response( |
| 312 | tweets=[self._tweet("1", "text", "u1", metrics=metrics)], |
| 313 | ) |
| 314 | items = xurl_x.parse_x_response(resp) |
| 315 | self.assertEqual(items[0]["engagement"]["likes"], 42) |
| 316 | self.assertEqual(items[0]["engagement"]["reposts"], 10) |
| 317 | self.assertEqual(items[0]["engagement"]["replies"], 5) |
| 318 | self.assertEqual(items[0]["engagement"]["quotes"], 2) |
| 319 | |
| 320 | def test_engagement_none_when_no_metrics(self): |
| 321 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 322 | items = xurl_x.parse_x_response(resp) |
| 323 | self.assertIsNone(items[0]["engagement"]) |
| 324 | |
| 325 | def test_text_truncated_to_500_chars(self): |
| 326 | long_text = "x" * 600 |
| 327 | resp = _make_api_response(tweets=[self._tweet("1", long_text, "u1")]) |
| 328 | items = xurl_x.parse_x_response(resp) |
| 329 | self.assertLessEqual(len(items[0]["text"]), 500) |
| 330 | |
| 331 | def test_id_prefixed_with_xurl(self): |
| 332 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 333 | items = xurl_x.parse_x_response(resp) |
| 334 | self.assertTrue(items[0]["id"].startswith("XURL")) |
| 335 | |
| 336 | def test_relevance_computed_when_topic_given(self): |
| 337 | resp = _make_api_response( |
| 338 | tweets=[self._tweet("1", "Claude Code is great for AI coding", "u1")], |
| 339 | ) |
| 340 | items = xurl_x.parse_x_response(resp, topic="Claude Code") |
| 341 | self.assertGreater(items[0]["relevance"], 0.5) |
| 342 | |
| 343 | def test_relevance_neutral_when_no_topic(self): |
| 344 | resp = _make_api_response(tweets=[self._tweet("1", "some text", "u1")]) |
| 345 | items = xurl_x.parse_x_response(resp) |
| 346 | self.assertEqual(items[0]["relevance"], 0.5) |
| 347 | |
| 348 | def test_url_empty_when_no_username(self): |
| 349 | # author_id not in includes.users → username="" |
| 350 | resp = _make_api_response(tweets=[self._tweet("999", "text", "unknown_uid")]) |
| 351 | items = xurl_x.parse_x_response(resp) |
| 352 | self.assertEqual(items[0]["url"], "") |
| 353 | |
| 354 | def test_multiple_tweets_parsed(self): |
| 355 | tweets = [self._tweet(str(i), f"tweet {i}", "u1") for i in range(5)] |
| 356 | resp = _make_api_response(tweets=tweets, users=[self._user("u1", "bob")]) |
| 357 | items = xurl_x.parse_x_response(resp) |
| 358 | self.assertEqual(len(items), 5) |
| 359 | |
| 360 | def test_empty_data_list(self): |
| 361 | resp = _make_api_response(tweets=[]) |
| 362 | self.assertEqual(xurl_x.parse_x_response(resp), []) |
| 363 | |
| 364 | def test_why_relevant_is_empty_string(self): |
| 365 | # xurl doesn't provide LLM-generated why_relevant (unlike xai_x) |
| 366 | resp = _make_api_response(tweets=[self._tweet("1", "text", "u1")]) |
| 367 | items = xurl_x.parse_x_response(resp) |
| 368 | self.assertEqual(items[0]["why_relevant"], "") |
| 369 | |
| 370 | # --------------------------------------------------------------------------- |
| 371 | # DEPTH_CONFIG |
| 372 | # --------------------------------------------------------------------------- |
| 373 | |
| 374 | |
| 375 | class TestDepthConfig(unittest.TestCase): |
| 376 | def test_all_standard_depths_present(self): |
| 377 | for depth in ("quick", "default", "deep"): |
| 378 | self.assertIn(depth, xurl_x.DEPTH_CONFIG) |
| 379 | |
| 380 | def test_deep_greater_than_quick(self): |
| 381 | self.assertGreater( |
| 382 | xurl_x.DEPTH_CONFIG["deep"], |
| 383 | xurl_x.DEPTH_CONFIG["quick"], |
| 384 | ) |
| 385 | |
| 386 | if __name__ == "__main__": |
| 387 | unittest.main() |
| 388 |