| 1 | import unittest |
| 2 | |
| 3 | from lib.tiktok import _parse_items |
| 4 | |
| 5 | |
| 6 | class TestTikTokAuthorTypeSafety(unittest.TestCase): |
| 7 | def _make_raw(self, **overrides): |
| 8 | base = { |
| 9 | "aweme_id": "1", |
| 10 | "desc": "test video", |
| 11 | "share_url": "https://www.tiktok.com/@u/video/1", |
| 12 | "author": {"unique_id": "testuser"}, |
| 13 | "statistics": {"play_count": 100, "digg_count": 50, "comment_count": 10, "share_count": 5}, |
| 14 | } |
| 15 | base.update(overrides) |
| 16 | return base |
| 17 | |
| 18 | def test_author_as_dict(self): |
| 19 | items = _parse_items([self._make_raw()], "test") |
| 20 | self.assertEqual("testuser", items[0]["author_name"]) |
| 21 | |
| 22 | def test_author_as_string(self): |
| 23 | items = _parse_items([self._make_raw(author="stringuser")], "test") |
| 24 | self.assertEqual("stringuser", items[0]["author_name"]) |
| 25 | |
| 26 | def test_author_missing(self): |
| 27 | raw = self._make_raw() |
| 28 | del raw["author"] |
| 29 | items = _parse_items([raw], "test") |
| 30 | self.assertEqual("", items[0]["author_name"]) |
| 31 | |
| 32 | def test_author_none(self): |
| 33 | items = _parse_items([self._make_raw(author=None)], "test") |
| 34 | self.assertEqual("", items[0]["author_name"]) |
| 35 | |
| 36 | |
| 37 | class TestTikTokStatsZeroPreserved(unittest.TestCase): |
| 38 | def test_zero_play_count(self): |
| 39 | raw = { |
| 40 | "aweme_id": "1", |
| 41 | "desc": "test", |
| 42 | "share_url": "https://www.tiktok.com/@u/video/1", |
| 43 | "author": {"unique_id": "u"}, |
| 44 | "statistics": {"play_count": 0, "digg_count": 0, "comment_count": 0, "share_count": 0}, |
| 45 | } |
| 46 | items = _parse_items([raw], "test") |
| 47 | self.assertEqual(0, items[0]["engagement"]["views"]) |
| 48 | self.assertEqual(0, items[0]["engagement"]["likes"]) |
| 49 | self.assertEqual(0, items[0]["engagement"]["comments"]) |
| 50 | self.assertEqual(0, items[0]["engagement"]["shares"]) |
| 51 | |
| 52 | def test_stats_missing(self): |
| 53 | raw = { |
| 54 | "aweme_id": "1", |
| 55 | "desc": "test", |
| 56 | "share_url": "https://www.tiktok.com/@u/video/1", |
| 57 | "author": {"unique_id": "u"}, |
| 58 | } |
| 59 | items = _parse_items([raw], "test") |
| 60 | self.assertEqual(0, items[0]["engagement"]["views"]) |
| 61 | |
| 62 | def test_stats_as_non_dict(self): |
| 63 | raw = { |
| 64 | "aweme_id": "1", |
| 65 | "desc": "test", |
| 66 | "share_url": "https://www.tiktok.com/@u/video/1", |
| 67 | "author": {"unique_id": "u"}, |
| 68 | "statistics": "invalid", |
| 69 | } |
| 70 | items = _parse_items([raw], "test") |
| 71 | self.assertEqual(0, items[0]["engagement"]["views"]) |
| 72 | |
| 73 | |
| 74 | class TestExpandTikTokQueries(unittest.TestCase): |
| 75 | """Tests for expand_tiktok_queries() multi-query generation.""" |
| 76 | |
| 77 | def test_default_depth_returns_two_plus_queries(self): |
| 78 | from lib.tiktok import expand_tiktok_queries |
| 79 | queries = expand_tiktok_queries("Kanye West", "default") |
| 80 | self.assertGreaterEqual(len(queries), 2) |
| 81 | # Breaking_news intent should include reaction/edit variant |
| 82 | variant_found = any( |
| 83 | "reaction" in q.lower() or "edit" in q.lower() or "trend" in q.lower() |
| 84 | for q in queries |
| 85 | ) |
| 86 | self.assertTrue(variant_found, f"Expected reaction/edit/trend variant: {queries}") |
| 87 | |
| 88 | def test_product_intent_includes_review_variant(self): |
| 89 | from lib.tiktok import expand_tiktok_queries |
| 90 | # "best laptop for coding" triggers the product intent (best .* for pattern) |
| 91 | queries = expand_tiktok_queries("best laptop for coding", "deep") |
| 92 | variant_found = any( |
| 93 | "review" in q.lower() or "haul" in q.lower() or "unboxing" in q.lower() |
| 94 | for q in queries |
| 95 | ) |
| 96 | self.assertTrue(variant_found, f"Expected review/haul/unboxing variant: {queries}") |
| 97 | |
| 98 | def test_quick_depth_returns_one_query(self): |
| 99 | from lib.tiktok import expand_tiktok_queries |
| 100 | queries = expand_tiktok_queries("Kanye West", "quick") |
| 101 | self.assertEqual(len(queries), 1) |
| 102 | |
| 103 | |
| 104 | class TestTikTokCommentsGate(unittest.TestCase): |
| 105 | def test_gate_requires_key_and_token(self): |
| 106 | from lib import env |
| 107 | self.assertFalse(env.is_tiktok_comments_available({})) |
| 108 | self.assertFalse(env.is_tiktok_comments_available( |
| 109 | {"SCRAPECREATORS_API_KEY": "k"} |
| 110 | )) |
| 111 | self.assertFalse(env.is_tiktok_comments_available( |
| 112 | {"INCLUDE_SOURCES": "tiktok_comments"} |
| 113 | )) |
| 114 | self.assertTrue(env.is_tiktok_comments_available( |
| 115 | {"SCRAPECREATORS_API_KEY": "k", "INCLUDE_SOURCES": "tiktok,tiktok_comments"} |
| 116 | )) |
| 117 | |
| 118 | def test_gate_case_matches_youtube_pattern(self): |
| 119 | from lib import env |
| 120 | # Matches the existing youtube_comments behaviour — plain substring match via _parse_include_sources. |
| 121 | self.assertTrue(env.is_tiktok_comments_available( |
| 122 | {"SCRAPECREATORS_API_KEY": "k", "INCLUDE_SOURCES": "TIKTOK,TIKTOK_COMMENTS"} |
| 123 | )) |
| 124 | |
| 125 | |
| 126 | class TestTikTokEnrichWithComments(unittest.TestCase): |
| 127 | def test_empty_items_returns_empty(self): |
| 128 | from lib import tiktok |
| 129 | self.assertEqual([], tiktok.enrich_with_comments([], token="k")) |
| 130 | |
| 131 | def test_missing_token_is_noop(self): |
| 132 | from lib import tiktok |
| 133 | items = [{"video_id": "1", "url": "https://www.tiktok.com/@u/video/1", "engagement": {"views": 100}}] |
| 134 | result = tiktok.enrich_with_comments(items, token="") |
| 135 | self.assertNotIn("top_comments", result[0]) |
| 136 | |
| 137 | def test_fetch_post_comments_parses_sc_response(self): |
| 138 | from unittest.mock import patch |
| 139 | from lib import tiktok |
| 140 | |
| 141 | fake_sc_response = { |
| 142 | "comments": [ |
| 143 | {"text": "loved it", "user": {"nickname": "Alice"}, |
| 144 | "digg_count": 420, "create_time": 1709251200}, |
| 145 | {"text": "meh", "user": {"nickname": "Bob"}, |
| 146 | "digg_count": 3, "create_time": 1709251300}, |
| 147 | {"text": "", "user": {"nickname": "Skip"}, |
| 148 | "digg_count": 999, "create_time": 1709251400}, |
| 149 | ], |
| 150 | "total": 3, |
| 151 | } |
| 152 | |
| 153 | with patch.object(tiktok.http, "get", return_value=fake_sc_response): |
| 154 | out = tiktok._fetch_post_comments( |
| 155 | "https://www.tiktok.com/@u/video/1", |
| 156 | token="k", |
| 157 | max_comments=5, |
| 158 | ) |
| 159 | # Empty-text comment dropped; rest sorted desc by digg_count. |
| 160 | self.assertEqual(2, len(out)) |
| 161 | self.assertEqual("loved it", out[0]["text"]) |
| 162 | self.assertEqual(420, out[0]["digg_count"]) |
| 163 | self.assertEqual("Alice", out[0]["author"]) |
| 164 | self.assertEqual("2024-03-01", out[0]["date"]) |
| 165 | self.assertEqual(3, out[1]["digg_count"]) |
| 166 | |
| 167 | def test_fetch_post_comments_prefers_unique_id_over_nickname(self): |
| 168 | """Author prefers unique_id (@handle) over nickname (display name).""" |
| 169 | from unittest.mock import patch |
| 170 | from lib import tiktok |
| 171 | |
| 172 | fake_sc_response = { |
| 173 | "comments": [ |
| 174 | {"text": "first", "user": {"unique_id": "moosanoormahomed", "nickname": "Moosa Noormahomed"}, |
| 175 | "digg_count": 3986, "create_time": 1709251200}, |
| 176 | {"text": "second", "user": {"nickname": "Muna9e"}, # no unique_id, falls back to nickname |
| 177 | "digg_count": 925, "create_time": 1709251300}, |
| 178 | {"text": "third", "user": {}, # neither - empty string |
| 179 | "digg_count": 100, "create_time": 1709251400}, |
| 180 | ], |
| 181 | "total": 3, |
| 182 | } |
| 183 | |
| 184 | with patch.object(tiktok.http, "get", return_value=fake_sc_response): |
| 185 | out = tiktok._fetch_post_comments( |
| 186 | "https://www.tiktok.com/@u/video/1", |
| 187 | token="k", |
| 188 | max_comments=5, |
| 189 | ) |
| 190 | self.assertEqual(3, len(out)) |
| 191 | # unique_id wins over nickname when both present |
| 192 | self.assertEqual("moosanoormahomed", out[0]["author"]) |
| 193 | # nickname used when unique_id missing |
| 194 | self.assertEqual("Muna9e", out[1]["author"]) |
| 195 | # both missing → empty string, comment still included |
| 196 | self.assertEqual("", out[2]["author"]) |
| 197 | |
| 198 | def test_fetch_post_comments_swallows_http_error(self): |
| 199 | from unittest.mock import patch |
| 200 | from lib import tiktok |
| 201 | |
| 202 | with patch.object(tiktok.http, "get", side_effect=Exception("429 rate limit")): |
| 203 | out = tiktok._fetch_post_comments( |
| 204 | "https://www.tiktok.com/@u/video/1", |
| 205 | token="k", |
| 206 | max_comments=5, |
| 207 | ) |
| 208 | self.assertEqual([], out) |
| 209 | |
| 210 | def test_enrich_attaches_top_comments_to_top_ranked_items(self): |
| 211 | from unittest.mock import patch |
| 212 | from lib import tiktok |
| 213 | |
| 214 | items = [ |
| 215 | {"video_id": "low", "url": "https://www.tiktok.com/@u/video/low", |
| 216 | "engagement": {"views": 10, "likes": 1, "comments": 0}}, |
| 217 | {"video_id": "high", "url": "https://www.tiktok.com/@u/video/high", |
| 218 | "engagement": {"views": 10000, "likes": 500, "comments": 30}}, |
| 219 | {"video_id": "mid", "url": "https://www.tiktok.com/@u/video/mid", |
| 220 | "engagement": {"views": 1000, "likes": 50, "comments": 5}}, |
| 221 | ] |
| 222 | with patch.object(tiktok, "_fetch_post_comments") as mock_fetch: |
| 223 | mock_fetch.return_value = [ |
| 224 | {"author": "A", "text": "fire", "digg_count": 100, "date": "2024-03-01"} |
| 225 | ] |
| 226 | tiktok.enrich_with_comments(items, token="k", max_posts=2) |
| 227 | # High and mid get comments; low does not. |
| 228 | by_id = {i["video_id"]: i for i in items} |
| 229 | self.assertIn("top_comments", by_id["high"]) |
| 230 | self.assertIn("top_comments", by_id["mid"]) |
| 231 | self.assertNotIn("top_comments", by_id["low"]) |
| 232 | |
| 233 | if __name__ == "__main__": |
| 234 | unittest.main() |
| 235 |