import json import sys import unittest from pathlib import Path 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, ) 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, ), ) if __name__ == "__main__": unittest.main()