| 1 | """Tests for bluesky module.""" |
| 2 | |
| 3 | import os |
| 4 | import unittest |
| 5 | from unittest.mock import patch, MagicMock |
| 6 | |
| 7 | from lib import bluesky |
| 8 | |
| 9 | |
| 10 | class TestExtractCoreSubject(unittest.TestCase): |
| 11 | def test_strips_prefix(self): |
| 12 | result = bluesky._extract_core_subject("what are people saying about claude code") |
| 13 | self.assertEqual(result, "claude code") |
| 14 | |
| 15 | def test_strips_noise(self): |
| 16 | result = bluesky._extract_core_subject("latest trending news claude code") |
| 17 | self.assertNotIn("latest", result) |
| 18 | self.assertNotIn("trending", result) |
| 19 | self.assertIn("claude", result) |
| 20 | |
| 21 | def test_preserves_core(self): |
| 22 | result = bluesky._extract_core_subject("react native") |
| 23 | self.assertEqual(result, "react native") |
| 24 | |
| 25 | |
| 26 | class TestParseDate(unittest.TestCase): |
| 27 | def test_indexed_at_iso(self): |
| 28 | item = {"indexedAt": "2024-06-15T12:00:00Z"} |
| 29 | self.assertEqual(bluesky._parse_date(item), "2024-06-15") |
| 30 | |
| 31 | def test_created_at_iso(self): |
| 32 | item = {"createdAt": "2024-03-01T08:30:00.000Z"} |
| 33 | self.assertEqual(bluesky._parse_date(item), "2024-03-01") |
| 34 | |
| 35 | def test_indexed_at_preferred_over_created_at(self): |
| 36 | item = {"indexedAt": "2024-06-15T12:00:00Z", "createdAt": "2024-06-14T12:00:00Z"} |
| 37 | self.assertEqual(bluesky._parse_date(item), "2024-06-15") |
| 38 | |
| 39 | def test_none_returns_none(self): |
| 40 | self.assertIsNone(bluesky._parse_date({})) |
| 41 | |
| 42 | def test_invalid_date_returns_none(self): |
| 43 | self.assertIsNone(bluesky._parse_date({"indexedAt": "not-a-date"})) |
| 44 | |
| 45 | |
| 46 | class TestParseBlueskyResponse(unittest.TestCase): |
| 47 | def test_basic_post(self): |
| 48 | response = { |
| 49 | "posts": [{ |
| 50 | "uri": "at://did:plc:abc123/app.bsky.feed.post/xyz789", |
| 51 | "author": {"handle": "alice.bsky.social", "displayName": "Alice"}, |
| 52 | "record": {"text": "Hello world", "createdAt": "2024-06-15T12:00:00Z"}, |
| 53 | "indexedAt": "2024-06-15T12:01:00Z", |
| 54 | "likeCount": 10, |
| 55 | "repostCount": 5, |
| 56 | "replyCount": 3, |
| 57 | "quoteCount": 1, |
| 58 | }] |
| 59 | } |
| 60 | items = bluesky.parse_bluesky_response(response) |
| 61 | self.assertEqual(len(items), 1) |
| 62 | self.assertEqual(items[0]["handle"], "alice.bsky.social") |
| 63 | self.assertEqual(items[0]["display_name"], "Alice") |
| 64 | self.assertEqual(items[0]["text"], "Hello world") |
| 65 | self.assertEqual(items[0]["url"], "https://bsky.app/profile/alice.bsky.social/post/xyz789") |
| 66 | self.assertEqual(items[0]["engagement"]["likes"], 10) |
| 67 | self.assertEqual(items[0]["engagement"]["reposts"], 5) |
| 68 | self.assertEqual(items[0]["date"], "2024-06-15") |
| 69 | |
| 70 | def test_empty_response(self): |
| 71 | items = bluesky.parse_bluesky_response({}) |
| 72 | self.assertEqual(items, []) |
| 73 | |
| 74 | def test_missing_fields(self): |
| 75 | response = {"posts": [{"uri": "", "author": {}, "record": {}}]} |
| 76 | items = bluesky.parse_bluesky_response(response) |
| 77 | self.assertEqual(len(items), 1) |
| 78 | self.assertEqual(items[0]["handle"], "") |
| 79 | self.assertEqual(items[0]["text"], "") |
| 80 | |
| 81 | def test_relevance_decreases_with_position(self): |
| 82 | response = {"posts": [ |
| 83 | {"uri": f"at://did/app.bsky.feed.post/{i}", "author": {"handle": f"u{i}"}, "record": {"text": f"post {i}"}} |
| 84 | for i in range(5) |
| 85 | ]} |
| 86 | items = bluesky.parse_bluesky_response(response) |
| 87 | self.assertGreater(items[0]["relevance"], items[4]["relevance"]) |
| 88 | |
| 89 | |
| 90 | class TestDepthConfig(unittest.TestCase): |
| 91 | def test_all_depths_exist(self): |
| 92 | for depth in ("quick", "default", "deep"): |
| 93 | self.assertIn(depth, bluesky.DEPTH_CONFIG) |
| 94 | |
| 95 | def test_deep_has_more_results(self): |
| 96 | quick = bluesky.DEPTH_CONFIG["quick"] |
| 97 | deep = bluesky.DEPTH_CONFIG["deep"] |
| 98 | self.assertGreater(deep, quick) |
| 99 | |
| 100 | |
| 101 | class TestCreateSession(unittest.TestCase): |
| 102 | def setUp(self): |
| 103 | bluesky._cached_token = None |
| 104 | |
| 105 | def tearDown(self): |
| 106 | bluesky._cached_token = None |
| 107 | |
| 108 | @patch("lib.bluesky.http.request") |
| 109 | def test_returns_token(self, mock_request): |
| 110 | mock_request.return_value = {"accessJwt": "tok123", "refreshJwt": "ref456"} |
| 111 | token = bluesky._create_session("user.bsky.social", "app-pw") |
| 112 | self.assertEqual(token, "tok123") |
| 113 | mock_request.assert_called_once() |
| 114 | |
| 115 | @patch("lib.bluesky.http.request") |
| 116 | def test_caches_token(self, mock_request): |
| 117 | mock_request.return_value = {"accessJwt": "tok123", "refreshJwt": "ref456"} |
| 118 | bluesky._create_session("user.bsky.social", "app-pw") |
| 119 | bluesky._create_session("user.bsky.social", "app-pw") |
| 120 | mock_request.assert_called_once() # Only one HTTP call |
| 121 | |
| 122 | @patch("lib.bluesky.http.request") |
| 123 | def test_returns_none_on_failure(self, mock_request): |
| 124 | mock_request.side_effect = Exception("connection refused") |
| 125 | token = bluesky._create_session("user.bsky.social", "app-pw") |
| 126 | self.assertIsNone(token) |
| 127 | self.assertIn("connection refused", bluesky._session_error) |
| 128 | |
| 129 | @patch("lib.bluesky.http.request") |
| 130 | def test_returns_none_on_missing_jwt(self, mock_request): |
| 131 | mock_request.return_value = {"did": "did:plc:abc"} |
| 132 | token = bluesky._create_session("user.bsky.social", "app-pw") |
| 133 | self.assertIsNone(token) |
| 134 | |
| 135 | |
| 136 | class TestSearchBlueskyAuth(unittest.TestCase): |
| 137 | def setUp(self): |
| 138 | bluesky._cached_token = None |
| 139 | |
| 140 | def tearDown(self): |
| 141 | bluesky._cached_token = None |
| 142 | |
| 143 | def test_no_config_returns_error(self): |
| 144 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09") |
| 145 | self.assertEqual(result["posts"], []) |
| 146 | self.assertIn("not configured", result["error"]) |
| 147 | |
| 148 | def test_empty_config_returns_error(self): |
| 149 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config={}) |
| 150 | self.assertEqual(result["posts"], []) |
| 151 | self.assertIn("not configured", result["error"]) |
| 152 | |
| 153 | @patch("lib.bluesky.http.request") |
| 154 | def test_auth_failure_returns_specific_error(self, mock_request): |
| 155 | mock_request.side_effect = Exception("connection refused") |
| 156 | config = {"BSKY_HANDLE": "user.bsky.social", "BSKY_APP_PASSWORD": "pw"} |
| 157 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config=config) |
| 158 | self.assertEqual(result["posts"], []) |
| 159 | self.assertIn("connection refused", result["error"]) |
| 160 | self.assertNotIn("auth failed", result["error"]) |
| 161 | |
| 162 | @patch("lib.bluesky.http.request") |
| 163 | def test_cloudflare_403_returns_network_error(self, mock_request): |
| 164 | from lib.http import HTTPError |
| 165 | mock_request.side_effect = HTTPError("HTTP 403: Forbidden", 403, "<html>Cloudflare</html>") |
| 166 | config = {"BSKY_HANDLE": "user.bsky.social", "BSKY_APP_PASSWORD": "pw"} |
| 167 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config=config) |
| 168 | self.assertEqual(result["posts"], []) |
| 169 | self.assertIn("Cloudflare", result["error"]) |
| 170 | self.assertIn("network", result["error"].lower()) |
| 171 | |
| 172 | @patch("lib.bluesky.http.request") |
| 173 | def test_401_returns_credentials_error(self, mock_request): |
| 174 | from lib.http import HTTPError |
| 175 | mock_request.side_effect = HTTPError("HTTP 401: Unauthorized", 401, "") |
| 176 | config = {"BSKY_HANDLE": "user.bsky.social", "BSKY_APP_PASSWORD": "pw"} |
| 177 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config=config) |
| 178 | self.assertEqual(result["posts"], []) |
| 179 | self.assertIn("Invalid credentials", result["error"]) |
| 180 | |
| 181 | @patch("lib.bluesky.http.request") |
| 182 | def test_successful_search_passes_bearer(self, mock_request): |
| 183 | # First call: createSession, second call: searchPosts |
| 184 | mock_request.side_effect = [ |
| 185 | {"accessJwt": "tok123", "refreshJwt": "ref456"}, |
| 186 | {"posts": [{"uri": "at://did/app.bsky.feed.post/abc", "author": {"handle": "u1"}, "record": {"text": "hi"}}]}, |
| 187 | ] |
| 188 | config = {"BSKY_HANDLE": "user.bsky.social", "BSKY_APP_PASSWORD": "pw"} |
| 189 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config=config) |
| 190 | self.assertEqual(len(result["posts"]), 1) |
| 191 | # Verify the search call included the Bearer token |
| 192 | search_call = mock_request.call_args_list[1] |
| 193 | self.assertEqual(search_call.kwargs.get("headers", {}), {"Authorization": "Bearer tok123"}) |
| 194 | |
| 195 | @patch("lib.bluesky.http.request") |
| 196 | def test_401_search_refreshes_session_once(self, mock_request): |
| 197 | from lib.http import HTTPError |
| 198 | |
| 199 | mock_request.side_effect = [ |
| 200 | {"accessJwt": "tok-old", "refreshJwt": "ref-old"}, |
| 201 | HTTPError("HTTP 401: Unauthorized", 401, ""), |
| 202 | {"accessJwt": "tok-new", "refreshJwt": "ref-new"}, |
| 203 | {"posts": [{"uri": "at://did/app.bsky.feed.post/abc", "author": {"handle": "u1"}, "record": {"text": "hi"}}]}, |
| 204 | ] |
| 205 | config = {"BSKY_HANDLE": "user.bsky.social", "BSKY_APP_PASSWORD": "pw"} |
| 206 | result = bluesky.search_bluesky("test", "2026-01-01", "2026-03-09", config=config) |
| 207 | self.assertEqual(len(result["posts"]), 1) |
| 208 | self.assertEqual(mock_request.call_count, 4) |
| 209 | self.assertEqual(mock_request.call_args_list[3].kwargs.get("headers", {}), {"Authorization": "Bearer tok-new"}) |
| 210 | |
| 211 | |
| 212 | class TestSearchEndpointHostResolution(unittest.TestCase): |
| 213 | """The default search host moved from `public.api.bsky.app` (the |
| 214 | unauthenticated public mirror, now BunnyCDN-blocked for searchPosts) to |
| 215 | `api.bsky.app` (the canonical authenticated AppView). BSKY_SEARCH_HOST |
| 216 | env var or config value can override the default if Bluesky migrates |
| 217 | infrastructure again. Same os.environ-or-config hybrid pattern as |
| 218 | LAST30DAYS_STORE. |
| 219 | """ |
| 220 | |
| 221 | def setUp(self): |
| 222 | # Snapshot env so per-test overrides don't leak |
| 223 | self._saved_env = os.environ.pop("BSKY_SEARCH_HOST", None) |
| 224 | |
| 225 | def tearDown(self): |
| 226 | if self._saved_env is not None: |
| 227 | os.environ["BSKY_SEARCH_HOST"] = self._saved_env |
| 228 | else: |
| 229 | os.environ.pop("BSKY_SEARCH_HOST", None) |
| 230 | |
| 231 | def test_resolver_default_uses_canonical_appview(self): |
| 232 | # Regression guard against the public mirror reappearing as the default. |
| 233 | # Anchored at the resolver because that is the code path search_bluesky |
| 234 | # actually calls; a module-level constant would not catch a resolver |
| 235 | # regression. |
| 236 | self.assertIn("api.bsky.app", bluesky._resolve_search_url()) |
| 237 | |
| 238 | def test_resolver_default_does_not_use_public_mirror(self): |
| 239 | # Hard regression guard — the exact host that BunnyCDN was blocking. |
| 240 | # Asserted at the resolver level (the runtime path) so a default-host |
| 241 | # regression in _resolve_search_url is actually caught. |
| 242 | self.assertNotIn("public.api.bsky.app", bluesky._resolve_search_url()) |
| 243 | |
| 244 | def test_resolver_default_when_no_override(self): |
| 245 | self.assertEqual( |
| 246 | bluesky._resolve_search_url(), |
| 247 | "https://api.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 248 | ) |
| 249 | |
| 250 | def test_resolver_env_var_override(self): |
| 251 | os.environ["BSKY_SEARCH_HOST"] = "staging.bsky.app" |
| 252 | self.assertEqual( |
| 253 | bluesky._resolve_search_url(), |
| 254 | "https://staging.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 255 | ) |
| 256 | |
| 257 | def test_resolver_config_dict_override(self): |
| 258 | # User has BSKY_SEARCH_HOST only in .env file (project loads .env into |
| 259 | # config, not os.environ). Resolver must read both. |
| 260 | url = bluesky._resolve_search_url({"BSKY_SEARCH_HOST": "pds.example.com"}) |
| 261 | self.assertEqual(url, "https://pds.example.com/xrpc/app.bsky.feed.searchPosts") |
| 262 | |
| 263 | def test_resolver_env_var_wins_over_config(self): |
| 264 | # When both are set, os.environ takes precedence (matches LAST30DAYS_STORE) |
| 265 | os.environ["BSKY_SEARCH_HOST"] = "shell-host.example" |
| 266 | url = bluesky._resolve_search_url({"BSKY_SEARCH_HOST": "config-host.example"}) |
| 267 | self.assertIn("shell-host.example", url) |
| 268 | self.assertNotIn("config-host.example", url) |
| 269 | |
| 270 | def test_resolver_output_does_not_use_public_mirror(self): |
| 271 | # Regression guard at the resolver level (not just the constant) — |
| 272 | # this is what runtime actually calls. The constant-level guard |
| 273 | # above doesn't catch a regression where the resolver reverts. |
| 274 | self.assertNotIn("public.api.bsky.app", bluesky._resolve_search_url()) |
| 275 | |
| 276 | def test_resolver_strips_surrounding_whitespace(self): |
| 277 | # Pre-fix: " api.bsky.app " produced "https:// api.bsky.app /xrpc/..." |
| 278 | # which urllib raises ValueError on with no hint the env var caused it. |
| 279 | os.environ["BSKY_SEARCH_HOST"] = " api.bsky.app " |
| 280 | self.assertEqual( |
| 281 | bluesky._resolve_search_url(), |
| 282 | "https://api.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 283 | ) |
| 284 | |
| 285 | def test_resolver_rejects_embedded_path(self): |
| 286 | # "my-proxy.com/xrpc/prefix" would have doubled the /xrpc/ segment. |
| 287 | # We fall back to the default to avoid a guaranteed 404. |
| 288 | os.environ["BSKY_SEARCH_HOST"] = "my-proxy.example.com/xrpc/prefix" |
| 289 | self.assertEqual( |
| 290 | bluesky._resolve_search_url(), |
| 291 | "https://api.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 292 | ) |
| 293 | |
| 294 | def test_resolver_strips_embedded_scheme(self): |
| 295 | # Users who paste a full URL get a sane outcome, not a malformed URL. |
| 296 | os.environ["BSKY_SEARCH_HOST"] = "https://api.bsky.app" |
| 297 | self.assertEqual( |
| 298 | bluesky._resolve_search_url(), |
| 299 | "https://api.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 300 | ) |
| 301 | |
| 302 | def test_resolver_empty_string_falls_back_to_default(self): |
| 303 | os.environ["BSKY_SEARCH_HOST"] = "" |
| 304 | self.assertEqual( |
| 305 | bluesky._resolve_search_url(), |
| 306 | "https://api.bsky.app/xrpc/app.bsky.feed.searchPosts", |
| 307 | ) |
| 308 | |
| 309 | |
| 310 | class TestAppPasswordFormat(unittest.TestCase): |
| 311 | """Bluesky app passwords are 19-char xxxx-xxxx-xxxx-xxxx (lowercase |
| 312 | alphanumeric, three hyphens at fixed positions). Main-account passwords |
| 313 | are accepted by createSession but are bad hygiene. The validator detects |
| 314 | the format mismatch without gating any caller. |
| 315 | """ |
| 316 | |
| 317 | def test_accepts_valid_app_password_form(self): |
| 318 | # Use a fake example — never a real password |
| 319 | self.assertTrue(bluesky._validate_app_password_format("wfwp-cq7o-5six-7wy5")) |
| 320 | |
| 321 | def test_rejects_length_15_string(self): |
| 322 | # The exact failure mode that triggered the 2026-05-04 investigation: |
| 323 | # user stored their main login password (15 chars) in BSKY_APP_PASSWORD |
| 324 | self.assertFalse(bluesky._validate_app_password_format("mainpassword123")) |
| 325 | |
| 326 | def test_rejects_16_char_no_hyphen_string(self): |
| 327 | # Hex-style API key shape — common confusion with other services |
| 328 | self.assertFalse(bluesky._validate_app_password_format("abcdef0123456789")) |
| 329 | |
| 330 | def test_rejects_uppercase_letters(self): |
| 331 | # Bluesky app passwords are all-lowercase by spec |
| 332 | self.assertFalse(bluesky._validate_app_password_format("WFWP-cq7o-5six-7wy5")) |
| 333 | |
| 334 | def test_rejects_underscore_separator(self): |
| 335 | # Wrong separator |
| 336 | self.assertFalse(bluesky._validate_app_password_format("wfwp_cq7o_5six_7wy5")) |
| 337 | |
| 338 | def test_rejects_special_chars_in_groups(self): |
| 339 | # Special characters are not part of the alphanumeric class |
| 340 | self.assertFalse(bluesky._validate_app_password_format("wfwp-cq7o-5six-7wy@")) |
| 341 | |
| 342 | def test_rejects_empty_string(self): |
| 343 | self.assertFalse(bluesky._validate_app_password_format("")) |
| 344 | |
| 345 | def test_rejects_none(self): |
| 346 | # Callers may pass config.get('BSKY_APP_PASSWORD') which is None when unset |
| 347 | self.assertFalse(bluesky._validate_app_password_format(None)) |
| 348 | |
| 349 | def test_rejects_integer(self): |
| 350 | # Defensive: don't crash if a numeric value sneaks in |
| 351 | self.assertFalse(bluesky._validate_app_password_format(123456789012345)) |
| 352 | |
| 353 | def test_rejects_list(self): |
| 354 | # Defensive: don't crash on iterables |
| 355 | self.assertFalse(bluesky._validate_app_password_format(["wfwp", "cq7o", "5six", "7wy5"])) |
| 356 | |
| 357 | if __name__ == "__main__": |
| 358 | unittest.main() |
| 359 |