250 lines
9.1 KiB
Python
250 lines
9.1 KiB
Python
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()
|