| 1 | import unittest |
| 2 | from unittest import mock |
| 3 | |
| 4 | from lib.linkedin import ( |
| 5 | _best_author_match, |
| 6 | _extract_posts, |
| 7 | _int_field, |
| 8 | _parse_date, |
| 9 | enrich_articles, |
| 10 | parse_linkedin_response, |
| 11 | parse_profile_articles, |
| 12 | search_linkedin, |
| 13 | ) |
| 14 | |
| 15 | |
| 16 | class TestParseLinkedinResponse(unittest.TestCase): |
| 17 | def _make_post(self, **overrides): |
| 18 | # Mirrors the live ScrapeCreators /v1/linkedin/search/posts post object: |
| 19 | # body in `description`, timestamp in `datePublished`, author as a nested |
| 20 | # dict, engagement in `likeCount`/`commentCount`, comments as a list. |
| 21 | base = { |
| 22 | "url": "https://www.linkedin.com/posts/example_123", |
| 23 | "datePublished": "2026-06-01T12:30:00.000Z", |
| 24 | "description": "Excited to share our latest product update.", |
| 25 | "author": { |
| 26 | "name": "Jane Doe", |
| 27 | "url": "https://www.linkedin.com/in/janedoe", |
| 28 | "followers": 1234, |
| 29 | }, |
| 30 | "comments": [{"author": "Bob", "text": "nice", "linkedinUrl": "x"}], |
| 31 | "likeCount": 42, |
| 32 | "commentCount": 5, |
| 33 | } |
| 34 | base.update(overrides) |
| 35 | return base |
| 36 | |
| 37 | def test_real_shape_post_parses_all_fields(self): |
| 38 | items = parse_linkedin_response({"posts": [self._make_post()]}) |
| 39 | self.assertEqual(1, len(items)) |
| 40 | item = items[0] |
| 41 | self.assertEqual("https://www.linkedin.com/posts/example_123", item["url"]) |
| 42 | self.assertEqual("Excited to share our latest product update.", item["text"]) |
| 43 | self.assertEqual("Jane Doe", item["author"]) |
| 44 | self.assertEqual("https://www.linkedin.com/in/janedoe", item["author_url"]) |
| 45 | self.assertEqual("2026-06-01", item["date"]) |
| 46 | self.assertEqual(42, item["engagement"]["likes"]) |
| 47 | self.assertEqual(5, item["engagement"]["comments"]) |
| 48 | self.assertFalse(item["is_article"]) |
| 49 | |
| 50 | def test_description_and_datepublished_only_parses_nonempty(self): |
| 51 | # Regression guard for the exact bug that shipped: the live API uses |
| 52 | # `description` + `datePublished`, and a parser keyed on `text`/`date` |
| 53 | # dropped every post (10 raw -> 0 items). This must stay non-empty. |
| 54 | raw = { |
| 55 | "description": "body via description field", |
| 56 | "datePublished": "2026-06-10T09:00:00Z", |
| 57 | "url": "https://www.linkedin.com/posts/abc", |
| 58 | } |
| 59 | items = parse_linkedin_response({"posts": [raw]}) |
| 60 | self.assertEqual(1, len(items)) |
| 61 | self.assertEqual("body via description field", items[0]["text"]) |
| 62 | self.assertEqual("2026-06-10", items[0]["date"]) |
| 63 | |
| 64 | def test_empty_posts_returns_empty_list(self): |
| 65 | self.assertEqual([], parse_linkedin_response({"posts": []})) |
| 66 | self.assertEqual([], parse_linkedin_response({})) |
| 67 | |
| 68 | def test_post_without_text_is_skipped(self): |
| 69 | raw = self._make_post() |
| 70 | del raw["description"] |
| 71 | items = parse_linkedin_response({"posts": [raw]}) |
| 72 | self.assertEqual([], items) |
| 73 | |
| 74 | def test_non_dict_post_is_skipped(self): |
| 75 | items = parse_linkedin_response({"posts": ["not-a-dict", None]}) |
| 76 | self.assertEqual([], items) |
| 77 | |
| 78 | def test_author_as_dict(self): |
| 79 | raw = self._make_post(author={"name": "Jane Doe", "full_name": "Jane M Doe"}) |
| 80 | items = parse_linkedin_response({"posts": [raw]}) |
| 81 | self.assertEqual("Jane Doe", items[0]["author"]) |
| 82 | |
| 83 | def test_author_dict_falls_back_to_full_name(self): |
| 84 | raw = self._make_post(author={"full_name": "Jane M Doe"}) |
| 85 | items = parse_linkedin_response({"posts": [raw]}) |
| 86 | self.assertEqual("Jane M Doe", items[0]["author"]) |
| 87 | |
| 88 | def test_author_missing_defaults_to_empty_string(self): |
| 89 | raw = self._make_post() |
| 90 | del raw["author"] |
| 91 | items = parse_linkedin_response({"posts": [raw]}) |
| 92 | self.assertEqual("", items[0]["author"]) |
| 93 | |
| 94 | def test_id_falls_back_to_generated_index(self): |
| 95 | # The live post object carries no top-level id/urn; the parser |
| 96 | # synthesizes a stable per-index id. |
| 97 | raw = self._make_post() |
| 98 | items = parse_linkedin_response({"posts": [raw]}) |
| 99 | self.assertEqual("LI1", items[0]["id"]) |
| 100 | |
| 101 | def test_explicit_urn_used_as_id(self): |
| 102 | raw = self._make_post(urn="urn:li:activity:999") |
| 103 | items = parse_linkedin_response({"posts": [raw]}) |
| 104 | self.assertEqual("urn:li:activity:999", items[0]["id"]) |
| 105 | |
| 106 | def test_alternate_field_names(self): |
| 107 | raw = { |
| 108 | "postId": "p1", |
| 109 | "content": "alternate field names", |
| 110 | "postUrl": "https://example.com/p1", |
| 111 | "authorName": "Alt Author", |
| 112 | "postedAt": "2026-05-15T10:00:00Z", |
| 113 | "likesCount": 10, |
| 114 | "commentsCount": 3, |
| 115 | "shares": 1, |
| 116 | } |
| 117 | items = parse_linkedin_response({"posts": [raw]}) |
| 118 | item = items[0] |
| 119 | self.assertEqual("p1", item["id"]) |
| 120 | self.assertEqual("alternate field names", item["text"]) |
| 121 | self.assertEqual("https://example.com/p1", item["url"]) |
| 122 | self.assertEqual("Alt Author", item["author"]) |
| 123 | self.assertEqual("2026-05-15", item["date"]) |
| 124 | self.assertEqual(10, item["engagement"]["likes"]) |
| 125 | self.assertEqual(3, item["engagement"]["comments"]) |
| 126 | self.assertEqual(1, item["engagement"]["reposts"]) |
| 127 | |
| 128 | |
| 129 | class TestExtractPosts(unittest.TestCase): |
| 130 | def test_extracts_from_posts_key(self): |
| 131 | self.assertEqual([{"a": 1}], _extract_posts({"posts": [{"a": 1}]})) |
| 132 | |
| 133 | def test_extracts_from_items_key(self): |
| 134 | self.assertEqual([{"a": 1}], _extract_posts({"items": [{"a": 1}]})) |
| 135 | |
| 136 | def test_extracts_from_data_key(self): |
| 137 | self.assertEqual([{"a": 1}], _extract_posts({"data": [{"a": 1}]})) |
| 138 | |
| 139 | def test_extracts_from_results_key(self): |
| 140 | self.assertEqual([{"a": 1}], _extract_posts({"results": [{"a": 1}]})) |
| 141 | |
| 142 | def test_non_dict_response_returns_empty(self): |
| 143 | self.assertEqual([], _extract_posts(["not", "a", "dict"])) |
| 144 | self.assertEqual([], _extract_posts(None)) |
| 145 | |
| 146 | def test_no_matching_key_returns_empty(self): |
| 147 | self.assertEqual([], _extract_posts({"unexpected": [1, 2, 3]})) |
| 148 | |
| 149 | |
| 150 | class TestParseDate(unittest.TestCase): |
| 151 | def test_extracts_date_from_iso_string(self): |
| 152 | self.assertEqual("2026-06-01", _parse_date("2026-06-01T12:30:00Z")) |
| 153 | |
| 154 | def test_plain_date_string(self): |
| 155 | self.assertEqual("2026-06-01", _parse_date("2026-06-01")) |
| 156 | |
| 157 | def test_none_returns_none(self): |
| 158 | self.assertIsNone(_parse_date(None)) |
| 159 | |
| 160 | def test_empty_string_returns_none(self): |
| 161 | self.assertIsNone(_parse_date("")) |
| 162 | |
| 163 | def test_no_date_pattern_returns_none(self): |
| 164 | self.assertIsNone(_parse_date("not a date")) |
| 165 | |
| 166 | |
| 167 | class TestIntField(unittest.TestCase): |
| 168 | def test_returns_first_present_key(self): |
| 169 | self.assertEqual(5, _int_field({"likes": 5, "likesCount": 10}, "likes", "likesCount")) |
| 170 | |
| 171 | def test_falls_back_to_second_key(self): |
| 172 | self.assertEqual(10, _int_field({"likesCount": 10}, "likes", "likesCount")) |
| 173 | |
| 174 | def test_missing_all_keys_returns_zero(self): |
| 175 | self.assertEqual(0, _int_field({}, "likes", "likesCount")) |
| 176 | |
| 177 | def test_non_numeric_value_returns_zero(self): |
| 178 | self.assertEqual(0, _int_field({"likes": "not-a-number"}, "likes")) |
| 179 | |
| 180 | def test_string_numeric_value_is_coerced(self): |
| 181 | self.assertEqual(7, _int_field({"likes": "7"}, "likes")) |
| 182 | |
| 183 | |
| 184 | class TestSearchLinkedin(unittest.TestCase): |
| 185 | def test_no_token_returns_empty_without_network_call(self): |
| 186 | with mock.patch("lib.linkedin.http.request") as mock_request: |
| 187 | result = search_linkedin("AI agents", "2026-05-27", "2026-06-26", token="") |
| 188 | self.assertEqual({"posts": []}, result) |
| 189 | mock_request.assert_not_called() |
| 190 | |
| 191 | def test_successful_search_returns_capped_posts(self): |
| 192 | raw_posts = [{"text": f"post {i}"} for i in range(20)] |
| 193 | with mock.patch( |
| 194 | "lib.linkedin.http.request", return_value={"posts": raw_posts} |
| 195 | ): |
| 196 | result = search_linkedin( |
| 197 | "AI agents", "2026-05-27", "2026-06-26", depth="quick", token="fake-token" |
| 198 | ) |
| 199 | # "quick" depth caps at 10 results per DEPTH_CONFIG |
| 200 | self.assertEqual(10, len(result["posts"])) |
| 201 | |
| 202 | def test_http_error_returns_empty_with_error_message(self): |
| 203 | from lib import http as http_module |
| 204 | |
| 205 | with mock.patch( |
| 206 | "lib.linkedin.http.request", |
| 207 | side_effect=http_module.HTTPError("rate limited", status_code=429), |
| 208 | ): |
| 209 | result = search_linkedin( |
| 210 | "AI agents", "2026-05-27", "2026-06-26", token="fake-token" |
| 211 | ) |
| 212 | self.assertEqual([], result["posts"]) |
| 213 | self.assertIn("error", result) |
| 214 | |
| 215 | |
| 216 | class TestDateRangeFiltering(unittest.TestCase): |
| 217 | def test_filters_out_posts_outside_range(self): |
| 218 | posts = [ |
| 219 | {"text": "in range", "date": "2026-06-15"}, |
| 220 | {"text": "too old", "date": "2026-01-01"}, |
| 221 | ] |
| 222 | items = parse_linkedin_response( |
| 223 | {"posts": posts}, from_date="2026-05-27", to_date="2026-06-26" |
| 224 | ) |
| 225 | self.assertEqual(1, len(items)) |
| 226 | self.assertEqual("in range", items[0]["text"]) |
| 227 | |
| 228 | def test_keeps_all_when_none_in_range(self): |
| 229 | # Graceful fallback: SC doesn't always return a parseable date, so if |
| 230 | # the filter would drop everything, keep the unfiltered set instead |
| 231 | # of returning zero results. |
| 232 | posts = [ |
| 233 | {"text": "old post", "date": "2026-01-01"}, |
| 234 | {"text": "older post", "date": "2025-12-01"}, |
| 235 | ] |
| 236 | items = parse_linkedin_response( |
| 237 | {"posts": posts}, from_date="2026-05-27", to_date="2026-06-26" |
| 238 | ) |
| 239 | self.assertEqual(2, len(items)) |
| 240 | |
| 241 | def test_no_filtering_when_dates_not_provided(self): |
| 242 | posts = [{"text": "any date", "date": "2020-01-01"}] |
| 243 | items = parse_linkedin_response({"posts": posts}) |
| 244 | self.assertEqual(1, len(items)) |
| 245 | |
| 246 | def test_posts_without_date_excluded_from_in_range_but_not_counted_as_failure(self): |
| 247 | posts = [ |
| 248 | {"text": "has date in range", "date": "2026-06-15"}, |
| 249 | {"text": "no date at all"}, |
| 250 | ] |
| 251 | items = parse_linkedin_response( |
| 252 | {"posts": posts}, from_date="2026-05-27", to_date="2026-06-26" |
| 253 | ) |
| 254 | # Only the dated, in-range post survives; the undated one isn't |
| 255 | # counted toward "in range" and gets dropped since in_range is non-empty. |
| 256 | self.assertEqual(1, len(items)) |
| 257 | self.assertEqual("has date in range", items[0]["text"]) |
| 258 | |
| 259 | |
| 260 | class TestArticleDetection(unittest.TestCase): |
| 261 | def test_pulse_url_tagged_as_article_and_boosted(self): |
| 262 | raw = { |
| 263 | "description": "long-form piece", |
| 264 | "datePublished": "2026-06-10", |
| 265 | "url": "https://www.linkedin.com/pulse/wtf-is-a-loop-matt-van-horn-abc", |
| 266 | } |
| 267 | item = parse_linkedin_response({"posts": [raw]})[0] |
| 268 | self.assertTrue(item["is_article"]) |
| 269 | self.assertEqual(0.9, item["relevance"]) |
| 270 | |
| 271 | def test_posts_url_not_article_default_relevance(self): |
| 272 | raw = { |
| 273 | "description": "ordinary status update", |
| 274 | "datePublished": "2026-06-10", |
| 275 | "url": "https://www.linkedin.com/posts/example_123", |
| 276 | } |
| 277 | item = parse_linkedin_response({"posts": [raw]})[0] |
| 278 | self.assertFalse(item["is_article"]) |
| 279 | self.assertEqual(0.5, item["relevance"]) |
| 280 | |
| 281 | |
| 282 | class TestProfileArticles(unittest.TestCase): |
| 283 | def _profile(self, **overrides): |
| 284 | base = { |
| 285 | "name": "Matt Van Horn", |
| 286 | "articles": [ |
| 287 | { |
| 288 | "headline": "WTF Is a Loop? Part 2", |
| 289 | "url": "https://www.linkedin.com/pulse/wtf-loop-part-2-abc", |
| 290 | "datePublished": "2026-06-20T21:06:18.000+00:00", |
| 291 | "articleBody": "", |
| 292 | }, |
| 293 | { |
| 294 | "headline": "Every Agentic Engineering Hack I Know", |
| 295 | "url": "https://www.linkedin.com/pulse/every-hack-def", |
| 296 | "datePublished": "2026-06-04T02:31:36.000+00:00", |
| 297 | "articleBody": "", |
| 298 | }, |
| 299 | ], |
| 300 | } |
| 301 | base.update(overrides) |
| 302 | return base |
| 303 | |
| 304 | def test_articles_parse_as_high_signal(self): |
| 305 | items = parse_profile_articles(self._profile()) |
| 306 | self.assertEqual(2, len(items)) |
| 307 | for it in items: |
| 308 | self.assertTrue(it["is_article"]) |
| 309 | self.assertEqual(0.9, it["relevance"]) |
| 310 | self.assertEqual("Matt Van Horn", it["author"]) |
| 311 | self.assertEqual("WTF Is a Loop? Part 2", items[0]["text"]) |
| 312 | |
| 313 | def test_articles_respect_date_range(self): |
| 314 | items = parse_profile_articles( |
| 315 | self._profile(), from_date="2026-06-15", to_date="2026-06-26" |
| 316 | ) |
| 317 | self.assertEqual(1, len(items)) |
| 318 | self.assertEqual("WTF Is a Loop? Part 2", items[0]["text"]) |
| 319 | |
| 320 | def test_empty_or_missing_articles(self): |
| 321 | self.assertEqual([], parse_profile_articles({"name": "X"})) |
| 322 | self.assertEqual([], parse_profile_articles({"name": "X", "articles": []})) |
| 323 | |
| 324 | def test_article_without_headline_skipped(self): |
| 325 | prof = {"name": "X", "articles": [{"url": "u", "datePublished": "2026-06-10"}]} |
| 326 | self.assertEqual([], parse_profile_articles(prof)) |
| 327 | |
| 328 | |
| 329 | class TestBestAuthorMatch(unittest.TestCase): |
| 330 | def test_exact_person_topic_matches(self): |
| 331 | items = [{"author": "Matt Van Horn", "author_url": "u-mvh"}] |
| 332 | self.assertEqual("u-mvh", _best_author_match(items, "Matt Van Horn")) |
| 333 | |
| 334 | def test_name_contained_in_longer_topic(self): |
| 335 | items = [{"author": "Matt Van Horn", "author_url": "u-mvh"}] |
| 336 | self.assertEqual( |
| 337 | "u-mvh", _best_author_match(items, "what is Matt Van Horn building") |
| 338 | ) |
| 339 | |
| 340 | def test_single_word_topic_never_enriches(self): |
| 341 | # "AI" is one token — a keyword topic, not a person. |
| 342 | items = [{"author": "Daisuke Tanaka", "author_url": "u-dt"}] |
| 343 | self.assertEqual("", _best_author_match(items, "AI")) |
| 344 | |
| 345 | def test_no_substring_false_positive_across_token_boundaries(self): |
| 346 | # Regression for the Greptile finding: a topic token must not match |
| 347 | # inside an unrelated author's name. "ai" must not hit "daisuke". |
| 348 | items = [{"author": "Daisuke Tanaka", "author_url": "u-dt"}] |
| 349 | self.assertEqual("", _best_author_match(items, "AI agents")) |
| 350 | |
| 351 | def test_picks_matching_author_not_first(self): |
| 352 | items = [ |
| 353 | {"author": "Eric Siu", "author_url": "u-eric"}, |
| 354 | {"author": "Matt Van Horn", "author_url": "u-mvh"}, |
| 355 | ] |
| 356 | self.assertEqual("u-mvh", _best_author_match(items, "Matt Van Horn loops")) |
| 357 | |
| 358 | |
| 359 | class TestEnrichArticles(unittest.TestCase): |
| 360 | def _person_items(self): |
| 361 | # Parsed post items as enrich_articles receives them: a person-topic |
| 362 | # search returns posts authored by the subject. |
| 363 | return [ |
| 364 | {"author": "Eric Siu", "author_url": "https://www.linkedin.com/in/ericosiu"}, |
| 365 | {"author": "Matt Van Horn", "author_url": "https://www.linkedin.com/in/mattvanhorn"}, |
| 366 | ] |
| 367 | |
| 368 | def test_person_topic_enriches_via_one_profile_call(self): |
| 369 | profile = { |
| 370 | "name": "Matt Van Horn", |
| 371 | "articles": [ |
| 372 | {"headline": "WTF Is a Loop?", "url": "https://www.linkedin.com/pulse/a", "datePublished": "2026-06-20"}, |
| 373 | ], |
| 374 | } |
| 375 | with mock.patch( |
| 376 | "lib.linkedin.http.request", return_value=profile |
| 377 | ) as mock_request: |
| 378 | arts = enrich_articles( |
| 379 | self._person_items(), "Matt Van Horn", token="fake", |
| 380 | from_date="2026-05-27", to_date="2026-06-26", |
| 381 | ) |
| 382 | self.assertEqual(1, len(arts)) |
| 383 | self.assertTrue(arts[0]["is_article"]) |
| 384 | self.assertEqual(0.9, arts[0]["relevance"]) |
| 385 | # Bounded: exactly one profile call. |
| 386 | self.assertEqual(1, mock_request.call_count) |
| 387 | # And it fetched the matching author's profile, not the first author. |
| 388 | called_profile = mock_request.call_args.kwargs["params"]["url"] |
| 389 | self.assertIn("mattvanhorn", called_profile) |
| 390 | |
| 391 | def test_keyword_topic_makes_no_profile_call(self): |
| 392 | with mock.patch("lib.linkedin.http.request") as mock_request: |
| 393 | arts = enrich_articles( |
| 394 | self._person_items(), "AI agents", token="fake", |
| 395 | from_date="2026-05-27", to_date="2026-06-26", |
| 396 | ) |
| 397 | self.assertEqual([], arts) |
| 398 | mock_request.assert_not_called() |
| 399 | |
| 400 | def test_no_token_no_call(self): |
| 401 | with mock.patch("lib.linkedin.http.request") as mock_request: |
| 402 | arts = enrich_articles(self._person_items(), "Matt Van Horn", token="") |
| 403 | self.assertEqual([], arts) |
| 404 | mock_request.assert_not_called() |
| 405 | |
| 406 | def test_profile_error_returns_empty(self): |
| 407 | from lib import http as http_module |
| 408 | |
| 409 | with mock.patch( |
| 410 | "lib.linkedin.http.request", |
| 411 | side_effect=http_module.HTTPError("boom", status_code=500), |
| 412 | ): |
| 413 | arts = enrich_articles(self._person_items(), "Matt Van Horn", token="fake") |
| 414 | self.assertEqual([], arts) |
| 415 | |
| 416 | def test_no_author_url_no_call(self): |
| 417 | items = [{"author": "Matt Van Horn", "author_url": ""}] |
| 418 | with mock.patch("lib.linkedin.http.request") as mock_request: |
| 419 | arts = enrich_articles(items, "Matt Van Horn", token="fake") |
| 420 | self.assertEqual([], arts) |
| 421 | mock_request.assert_not_called() |
| 422 | |
| 423 | |
| 424 | if __name__ == "__main__": |
| 425 | unittest.main() |
| 426 |