93 lines
3.2 KiB
Python
93 lines
3.2 KiB
Python
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,
|
|
)
|
|
|
|
|
|
class MatchLiveStreamTest(unittest.TestCase):
|
|
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)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|