| 1 | """YouTube comments via yt-dlp: the free, keyless path. |
| 2 | |
| 3 | ScrapeCreators used to be the only way to get YouTube comments. yt-dlp already |
| 4 | powers YouTube search and transcripts here and can fetch comments too, so the |
| 5 | comment lane no longer needs a paid key. These tests lock in that yt-dlp is |
| 6 | preferred, that ScrapeCreators still works as a fallback, and that a missing |
| 7 | key is no longer fatal. |
| 8 | """ |
| 9 | |
| 10 | import json |
| 11 | import unittest |
| 12 | from unittest import mock |
| 13 | |
| 14 | from lib import env, youtube_yt |
| 15 | from lib.subproc import SubprocResult |
| 16 | |
| 17 | |
| 18 | def _ytdlp_payload(comments): |
| 19 | """A yt-dlp --dump-single-json blob carrying `comments`.""" |
| 20 | return json.dumps({"id": "abc123", "title": "vid", "comments": comments}) |
| 21 | |
| 22 | |
| 23 | # yt-dlp's real comment shape, as emitted by --write-comments. |
| 24 | _RAW = [ |
| 25 | { |
| 26 | "author": "@BestFlorin", |
| 27 | "text": "Trump said the Hormuz strait is open", |
| 28 | "like_count": 11, |
| 29 | "_time_text": "2 days ago", |
| 30 | }, |
| 31 | { |
| 32 | "author": "@princem4006", |
| 33 | "text": "The U.S. cannot be trusted here", |
| 34 | "like_count": 7, |
| 35 | "_time_text": "1 day ago", |
| 36 | }, |
| 37 | ] |
| 38 | |
| 39 | |
| 40 | class TestFetchViaYtdlp(unittest.TestCase): |
| 41 | def test_parses_ytdlp_comments_into_canonical_shape(self): |
| 42 | """yt-dlp's like_count/_time_text map onto the engine's likes/date.""" |
| 43 | result = SubprocResult(returncode=0, stdout=_ytdlp_payload(_RAW), stderr="") |
| 44 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 45 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=result): |
| 46 | got = youtube_yt._fetch_video_comments_ytdlp("abc123", max_comments=5) |
| 47 | |
| 48 | self.assertEqual(2, len(got)) |
| 49 | self.assertEqual( |
| 50 | { |
| 51 | "author": "@BestFlorin", |
| 52 | "text": "Trump said the Hormuz strait is open", |
| 53 | "likes": 11, |
| 54 | "date": "2 days ago", |
| 55 | }, |
| 56 | got[0], |
| 57 | ) |
| 58 | |
| 59 | def test_honors_max_comments(self): |
| 60 | result = SubprocResult(returncode=0, stdout=_ytdlp_payload(_RAW), stderr="") |
| 61 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 62 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=result): |
| 63 | got = youtube_yt._fetch_video_comments_ytdlp("abc123", max_comments=1) |
| 64 | |
| 65 | self.assertEqual(1, len(got)) |
| 66 | |
| 67 | def test_returns_empty_when_ytdlp_not_installed(self): |
| 68 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=False): |
| 69 | self.assertEqual([], youtube_yt._fetch_video_comments_ytdlp("abc123")) |
| 70 | |
| 71 | def test_returns_empty_on_ytdlp_failure(self): |
| 72 | """A non-zero exit is a fetch error, not an empty comment section.""" |
| 73 | result = SubprocResult(returncode=1, stdout="", stderr="boom") |
| 74 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 75 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=result): |
| 76 | self.assertEqual([], youtube_yt._fetch_video_comments_ytdlp("abc123")) |
| 77 | |
| 78 | def test_command_requests_top_sorted_comments(self): |
| 79 | """Lock the command: top-sort and the max_comments cap must be present, |
| 80 | or a refactor could silently return arbitrary (newest) comments.""" |
| 81 | result = SubprocResult(returncode=0, stdout=_ytdlp_payload(_RAW), stderr="") |
| 82 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 83 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=result) as run: |
| 84 | youtube_yt._fetch_video_comments_ytdlp("abc123", max_comments=4) |
| 85 | |
| 86 | cmd = run.call_args.args[0] |
| 87 | joined = " ".join(cmd) |
| 88 | self.assertIn("--write-comments", cmd) |
| 89 | self.assertIn("comment_sort=top", joined) |
| 90 | self.assertIn("max_comments=4", joined) |
| 91 | self.assertTrue(any("watch?v=abc123" in a for a in cmd)) |
| 92 | |
| 93 | |
| 94 | class TestBackendPreference(unittest.TestCase): |
| 95 | def test_prefers_ytdlp_and_never_calls_scrapecreators(self): |
| 96 | """The free path wins: no SC credit is spent when yt-dlp delivers.""" |
| 97 | with mock.patch.object( |
| 98 | youtube_yt, |
| 99 | "_ytdlp_comments_result", |
| 100 | return_value=([{"author": "a", "text": "t", "likes": 1, "date": ""}], True), |
| 101 | ), mock.patch.object(youtube_yt.http, "get") as sc_get: |
| 102 | got = youtube_yt._fetch_video_comments("abc123", token="sk-live", max_comments=5) |
| 103 | |
| 104 | self.assertEqual(1, len(got)) |
| 105 | sc_get.assert_not_called() |
| 106 | |
| 107 | def test_falls_back_to_scrapecreators_when_ytdlp_fails(self): |
| 108 | """SC remains the backstop when yt-dlp is missing or throttled.""" |
| 109 | sc_payload = {"comments": [{"text": "from SC", "author": {"name": "@x"}, "likes": 3}]} |
| 110 | with mock.patch.object(youtube_yt, "_ytdlp_comments_result", return_value=([], False)), \ |
| 111 | mock.patch.object(youtube_yt.http, "get", return_value=sc_payload) as sc_get: |
| 112 | got = youtube_yt._fetch_video_comments("abc123", token="sk-live", max_comments=5) |
| 113 | |
| 114 | sc_get.assert_called_once() |
| 115 | self.assertEqual("from SC", got[0]["text"]) |
| 116 | |
| 117 | def test_no_token_and_ytdlp_failure_yields_no_comments_without_calling_sc(self): |
| 118 | with mock.patch.object(youtube_yt, "_ytdlp_comments_result", return_value=([], False)), \ |
| 119 | mock.patch.object(youtube_yt.http, "get") as sc_get: |
| 120 | got = youtube_yt._fetch_video_comments("abc123", token="", max_comments=5) |
| 121 | |
| 122 | self.assertEqual([], got) |
| 123 | sc_get.assert_not_called() |
| 124 | |
| 125 | def test_no_sc_fallback_when_ytdlp_succeeds_with_zero_comments(self): |
| 126 | """A video that genuinely has no comments must not burn an SC credit. |
| 127 | yt-dlp exit 0 + empty comments is success, not a throttle to retry.""" |
| 128 | ok_empty = SubprocResult(returncode=0, stdout=_ytdlp_payload([]), stderr="") |
| 129 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 130 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=ok_empty), \ |
| 131 | mock.patch.object(youtube_yt.http, "get") as sc_get: |
| 132 | got = youtube_yt._fetch_video_comments("abc123", token="sk-live", max_comments=5) |
| 133 | |
| 134 | self.assertEqual([], got) |
| 135 | sc_get.assert_not_called() |
| 136 | |
| 137 | def test_sc_fallback_fires_when_ytdlp_actually_fails(self): |
| 138 | """A non-zero exit is a real failure -> SC backstop should still fire.""" |
| 139 | failed = SubprocResult(returncode=1, stdout="", stderr="throttled") |
| 140 | sc_payload = {"comments": [{"text": "from SC", "author": {"name": "@x"}, "likes": 3}]} |
| 141 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 142 | mock.patch.object(youtube_yt.subproc, "run_with_timeout", return_value=failed), \ |
| 143 | mock.patch.object(youtube_yt.http, "get", return_value=sc_payload) as sc_get: |
| 144 | got = youtube_yt._fetch_video_comments("abc123", token="sk-live", max_comments=5) |
| 145 | |
| 146 | sc_get.assert_called_once() |
| 147 | self.assertEqual("from SC", got[0]["text"]) |
| 148 | |
| 149 | |
| 150 | class TestEnrichWithoutKey(unittest.TestCase): |
| 151 | def test_enriches_with_empty_token_when_ytdlp_available(self): |
| 152 | """A missing ScrapeCreators key must no longer disable comments.""" |
| 153 | items = [{"video_id": "abc123", "engagement": {"views": 100}}] |
| 154 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=True), \ |
| 155 | mock.patch.object( |
| 156 | youtube_yt, |
| 157 | "_fetch_video_comments", |
| 158 | return_value=[{"author": "@a", "text": "hi", "likes": 2, "date": ""}], |
| 159 | ): |
| 160 | youtube_yt.enrich_with_comments(items, token="") |
| 161 | |
| 162 | self.assertEqual("hi", items[0]["top_comments"][0]["text"]) |
| 163 | |
| 164 | def test_noop_with_no_token_and_no_ytdlp(self): |
| 165 | items = [{"video_id": "abc123", "engagement": {"views": 100}}] |
| 166 | with mock.patch.object(youtube_yt, "is_ytdlp_installed", return_value=False): |
| 167 | youtube_yt.enrich_with_comments(items, token="") |
| 168 | |
| 169 | self.assertNotIn("top_comments", items[0]) |
| 170 | |
| 171 | |
| 172 | class TestAvailabilityGate(unittest.TestCase): |
| 173 | def test_available_without_sc_key_when_ytdlp_installed(self): |
| 174 | """Comments are free now, so no key and no opt-in should be required.""" |
| 175 | with mock.patch.object(env, "is_ytdlp_available", return_value=True): |
| 176 | self.assertTrue(env.is_youtube_comments_available({})) |
| 177 | |
| 178 | def test_sc_path_still_available_when_ytdlp_missing(self): |
| 179 | cfg = { |
| 180 | "SCRAPECREATORS_API_KEY": "sk-live", |
| 181 | "INCLUDE_SOURCES": "youtube_comments", |
| 182 | } |
| 183 | with mock.patch.object(env, "is_ytdlp_available", return_value=False): |
| 184 | self.assertTrue(env.is_youtube_comments_available(cfg)) |
| 185 | |
| 186 | def test_unavailable_with_no_ytdlp_and_no_key(self): |
| 187 | with mock.patch.object(env, "is_ytdlp_available", return_value=False): |
| 188 | self.assertFalse(env.is_youtube_comments_available({})) |
| 189 | |
| 190 | def test_exclude_sources_still_suppresses_the_free_path(self): |
| 191 | """Comments going default-on must not defeat the documented off-switch.""" |
| 192 | cfg = {"EXCLUDE_SOURCES": "youtube_comments"} |
| 193 | with mock.patch.object(env, "is_ytdlp_available", return_value=True): |
| 194 | self.assertFalse(env.is_youtube_comments_available(cfg)) |
| 195 | |
| 196 | |
| 197 | if __name__ == "__main__": |
| 198 | unittest.main() |
| 199 |