test: cover match live stream repository updates

docker/crawler/tests/test_match_live_stream.py

@@ -13,10 +13,47 @@ from match_live_stream import (
     _build_run_result,
     choose_next_fetch_at,
     parse_match_live_detail_html,
+    MatchLiveStreamRepository,
 )
 
 
+class FakeCursor:
+    def __init__(self, fetchall_result=None):
+        self.fetchall_result = list(fetchall_result or [])
+        self.executed = []
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def execute(self, sql, params):
+        self.executed.append((sql, params))
+
+    def fetchall(self):
+        return list(self.fetchall_result)
+
+
+class FakeConnection:
+    def __init__(self, fetchall_result=None):
+        self.cursor_instance = FakeCursor(fetchall_result=fetchall_result)
+        self.commit_count = 0
+
+    def cursor(self):
+        return self.cursor_instance
+
+    def commit(self):
+        self.commit_count += 1
+
+
 class MatchLiveStreamTest(unittest.TestCase):
+    def _build_repo(self, *, prefix="la_", fetchall_result=None):
+        repo = object.__new__(MatchLiveStreamRepository)
+        repo.prefix = prefix
+        repo.conn = FakeConnection(fetchall_result=fetchall_result)
+        return repo
+
     def test_parse_json_ncctrials_detail_extracts_four_urls(self):
         payload = {
             "code": 200,
@@ -87,6 +124,70 @@ class MatchLiveStreamTest(unittest.TestCase):
     def test_choose_next_fetch_at_uses_600s_for_existing_streams(self):
         self.assertEqual(choose_next_fetch_at(now_ts=1000, has_existing_stream=True), 1600)
 
+    def test_load_due_rows_orders_by_due_time_and_honors_limit(self):
+        expected_rows = [{"id": 12, "source_url": "https://yyzb1.tv/room/7988511"}]
+        repo = self._build_repo(fetchall_result=expected_rows)
+
+        rows = repo.load_due_rows(now_ts=1711111111, limit=7)
+
+        self.assertEqual(rows, expected_rows)
+        self.assertEqual(repo.conn.commit_count, 0)
+        self.assertEqual(len(repo.conn.cursor_instance.executed), 1)
+        sql, params = repo.conn.cursor_instance.executed[0]
+        self.assertIn("FROM `la_match_live`", sql)
+        self.assertIn("`next_fetch_at` <= %s", sql)
+        self.assertIn("ORDER BY `next_fetch_at` ASC, `id` ASC LIMIT %s", sql)
+        self.assertEqual(params, (1711111111, 7))
+
+    def test_mark_failed_only_updates_fetch_fields_and_preserves_stream_urls(self):
+        repo = self._build_repo()
+        error_message = "x" * 600
+
+        repo.mark_failed(row_id=88, error_message=error_message, now_ts=2000, has_existing_stream=True)
+
+        self.assertEqual(repo.conn.commit_count, 1)
+        self.assertEqual(len(repo.conn.cursor_instance.executed), 1)
+        sql, params = repo.conn.cursor_instance.executed[0]
+        self.assertIn("UPDATE `la_match_live` SET", sql)
+        self.assertIn("`fetch_status`=2", sql)
+        self.assertIn("`fetch_error`=%s, `last_fetch_at`=%s, `next_fetch_at`=%s, `update_time`=%s", sql)
+        self.assertNotIn("`play_url_flv`=%s", sql)
+        self.assertNotIn("`play_url_hd_flv`=%s", sql)
+        self.assertNotIn("`play_url_m3u8`=%s", sql)
+        self.assertNotIn("`play_url_hd_m3u8`=%s", sql)
+        self.assertEqual(params, ("x" * 500, 2000, 2600, 2000, 88))
+
+    def test_mark_success_updates_all_stream_urls_and_resets_fetch_fields(self):
+        repo = self._build_repo()
+        stream_urls = {
+            "play_url_flv": "http://stream/flv",
+            "play_url_hd_flv": "http://stream/hdflv",
+            "play_url_m3u8": "http://stream/m3u8",
+            "play_url_hd_m3u8": "http://stream/hdm3u8",
+        }
+
+        repo.mark_success(row_id=66, stream_urls=stream_urls, now_ts=3000)
+
+        self.assertEqual(repo.conn.commit_count, 1)
+        self.assertEqual(len(repo.conn.cursor_instance.executed), 1)
+        sql, params = repo.conn.cursor_instance.executed[0]
+        self.assertIn("UPDATE `la_match_live` SET", sql)
+        self.assertIn("`play_url_flv`=%s, `play_url_hd_flv`=%s, `play_url_m3u8`=%s, `play_url_hd_m3u8`=%s", sql)
+        self.assertIn("`fetch_status`=1, `fetch_error`='', `last_fetch_at`=%s, `next_fetch_at`=%s, `update_time`=%s", sql)
+        self.assertEqual(
+            params,
+            (
+                "http://stream/flv",
+                "http://stream/hdflv",
+                "http://stream/m3u8",
+                "http://stream/hdm3u8",
+                3000,
+                3600,
+                3000,
+                66,
+            ),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()