import json import sys import unittest from pathlib import Path from unittest import mock ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from match_live_stream import ( _build_detail_url, _build_run_result, choose_next_fetch_at, parse_match_live_detail_html, MatchLiveStreamRepository, run, ) 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, "msg": "ok", "data": { "stream": { "flv": "http://a/flv", "hdFlv": "http://a/hdflv", "m3u8": "http://a/m3u8", "hdM3u8": "http://a/hdm3u8", } }, } parsed = parse_match_live_detail_html(f"detail({json.dumps(payload, ensure_ascii=False)})") self.assertEqual(parsed["play_url_m3u8"], "http://a/m3u8") self.assertEqual(parsed["play_url_hd_m3u8"], "http://a/hdm3u8") self.assertEqual(parsed["play_url_flv"], "http://a/flv") self.assertEqual(parsed["play_url_hd_flv"], "http://a/hdflv") def test_parse_json_ncctrials_detail_accepts_trailing_semicolon(self): payload = { "code": 200, "msg": "ok", "data": { "stream": { "flv": "http://b/flv", "hdFlv": "http://b/hdflv", "m3u8": "http://b/m3u8", "hdM3u8": "http://b/hdm3u8", } }, } parsed = parse_match_live_detail_html(f"detail({json.dumps(payload, ensure_ascii=False)});") self.assertEqual(parsed["play_url_m3u8"], "http://b/m3u8") self.assertEqual(parsed["play_url_hd_m3u8"], "http://b/hdm3u8") self.assertEqual(parsed["play_url_flv"], "http://b/flv") self.assertEqual(parsed["play_url_hd_flv"], "http://b/hdflv") def test_build_detail_url_derives_room_id_from_source_url(self): detail_url = _build_detail_url( "https://yyzb1.tv/room/7988511?scheduleId=undefined", now_ts=123456, ) self.assertEqual( detail_url, "https://json.ncctrials.com/room/7988511/detail.json?v=123456", ) def test_build_run_result_exposes_top_level_error_for_partial_failure(self): result = _build_run_result(candidate_count=3, success_count=2, failed_count=1) self.assertFalse(result["success"]) self.assertEqual(result["candidate_count"], 3) self.assertEqual(result["saved_count"], 2) self.assertEqual(result["count"], 2) self.assertEqual(result["failed_count"], 1) self.assertIn("失败 1 条", result["summary_text"]) self.assertEqual(result["error"], "赛事直播流抓取存在失败线路: 失败 1 条") def test_choose_next_fetch_at_uses_60s_for_empty_streams(self): self.assertEqual(choose_next_fetch_at(now_ts=1000, has_existing_stream=False), 1060) 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, ), ) def test_run_does_not_skip_due_rows_when_invoked_by_docker_task_runner(self): class FakeRepo: def __init__(self): self.closed = False self.mark_success_calls = [] def load_due_rows(self, now_ts, limit=30): return [ { "id": 1, "source_url": "https://yyzb1.tv/room/190118?scheduleId=undefined", "play_url_flv": "", "play_url_hd_flv": "", "play_url_m3u8": "", "play_url_hd_m3u8": "", } ] def mark_success(self, row_id, stream_urls, now_ts): self.mark_success_calls.append((row_id, stream_urls, now_ts)) def mark_failed(self, row_id, error_message, now_ts, has_existing_stream): raise AssertionError("docker task runner path should not fall into mark_failed for this test") def close(self): self.closed = True class FakeLock: def acquire(self): return False def release(self): return None repo = FakeRepo() 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", } with mock.patch.dict("os.environ", {"DQD_RUNNING_UNDER_TASK_RUNNER": "1"}, clear=False): with mock.patch("match_live_stream.MatchLiveStreamRepository", return_value=repo): with mock.patch("match_live_stream.TaskFileLock", return_value=FakeLock()): with mock.patch("match_live_stream._fetch_stream_payload", return_value=stream_urls): result = run({"prefix": "la_"}) self.assertTrue(result["success"]) self.assertEqual(result["candidate_count"], 1) self.assertEqual(result["saved_count"], 1) self.assertEqual(len(repo.mark_success_calls), 1) self.assertTrue(repo.closed) if __name__ == "__main__": unittest.main()