226 lines
11 KiB
Python
226 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import asyncio
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import web
|
|
|
|
|
|
class _JsonRequest:
|
|
def __init__(self, payload: dict[str, object]):
|
|
self._payload = payload
|
|
|
|
async def json(self) -> dict[str, object]:
|
|
return self._payload
|
|
|
|
|
|
def _json_body(resp) -> dict[str, object]: # noqa: ANN001
|
|
return json.loads(resp.body)
|
|
|
|
|
|
class WebDynamicProcessesTests(unittest.TestCase):
|
|
def test_process_monitor_entries_include_saved_youtube_pro_channels(self) -> None:
|
|
channels = [{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube2__lane1"}]
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
entries = web.process_monitor_entries()
|
|
|
|
pm2_names = {row["pm2"] for row in entries}
|
|
self.assertIn("youtube2__lane1", pm2_names)
|
|
|
|
def test_process_monitor_entries_default_pm2_name_when_missing(self) -> None:
|
|
channels = [{"id": "lane2", "name": "Lane 2"}]
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
entries = web.process_monitor_entries()
|
|
|
|
pm2_names = {row["pm2"] for row in entries}
|
|
self.assertIn("youtube2__lane2", pm2_names)
|
|
|
|
def test_script_for_pm2_or_pro_accepts_custom_saved_pm2_name(self) -> None:
|
|
channels = [{"id": "lane3", "name": "Lane 3", "pm2Name": "custom-lane-three"}]
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
resolved = web.script_for_pm2_or_pro("custom-lane-three")
|
|
|
|
self.assertEqual(resolved, "youtube.py")
|
|
|
|
def test_process_monitor_entries_keep_static_youtube_label_for_lane_one(self) -> None:
|
|
channels = [{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube"}]
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
entries = web.process_monitor_entries()
|
|
|
|
youtube_row = next(row for row in entries if row["pm2"] == "youtube")
|
|
self.assertEqual(youtube_row["label"], "YouTube")
|
|
|
|
def test_process_monitor_entries_do_not_duplicate_static_pm2_rows(self) -> None:
|
|
channels = [
|
|
{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube"},
|
|
{"id": "lane2", "name": "Lane 2", "pm2Name": "youtube2__lane2"},
|
|
]
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
entries = web.process_monitor_entries()
|
|
|
|
self.assertEqual(sum(1 for row in entries if row["pm2"] == "youtube"), 1)
|
|
self.assertEqual(sum(1 for row in entries if row["pm2"] == "youtube2__lane2"), 1)
|
|
|
|
def test_web_module_does_not_repeat_process_entry_function_defs(self) -> None:
|
|
tree = ast.parse((web.BASE_DIR / "web.py").read_text(encoding="utf-8"))
|
|
defs: dict[str, list[int]] = {}
|
|
for node in tree.body:
|
|
if isinstance(node, ast.FunctionDef):
|
|
defs.setdefault(node.name, []).append(node.lineno)
|
|
|
|
self.assertEqual(len(defs.get("youtube_pro_process_entries", [])), 1)
|
|
self.assertEqual(len(defs.get("process_monitor_entries", [])), 1)
|
|
|
|
def test_status_marks_local_saved_process_as_configured_when_not_running(self) -> None:
|
|
original_use_pm2 = web.process_backend._use_pm2
|
|
web.process_backend._use_pm2 = False
|
|
try:
|
|
with patch.object(web.process_backend, "full_status_raw", AsyncMock(return_value="local table")):
|
|
with patch("web.get_process_info", AsyncMock(return_value={"process_status": "not_found", "out_path": None, "err_path": None})):
|
|
with patch("web.read_log_path_normalized", AsyncMock(return_value="")):
|
|
with patch.object(web.process_backend, "live_ffmpeg_pids", AsyncMock(return_value=[])):
|
|
response = asyncio.run(web.status("youtube"))
|
|
finally:
|
|
web.process_backend._use_pm2 = original_use_pm2
|
|
|
|
payload = json.loads(response.body)
|
|
self.assertEqual(payload["process_status"], "configured")
|
|
self.assertEqual(payload["business_status"], "ready")
|
|
self.assertIn("已在控制台配置", payload["recent_log"])
|
|
self.assertTrue(payload["raw_status"].startswith("进程「youtube」已在控制台配置"))
|
|
|
|
def test_process_monitor_includes_runtime_snapshot_for_local_configured_process(self) -> None:
|
|
original_use_pm2 = web.process_backend._use_pm2
|
|
web.process_backend._use_pm2 = False
|
|
try:
|
|
with patch.object(web.process_backend, "ensure_mode", AsyncMock()):
|
|
with patch.object(
|
|
web.process_backend,
|
|
"process_infos",
|
|
AsyncMock(return_value={"youtube": {"process_status": "not_found", "out_path": None, "err_path": None}}),
|
|
):
|
|
with patch("web.read_log_path_normalized", AsyncMock(return_value="")):
|
|
payload = asyncio.run(web.process_monitor())
|
|
finally:
|
|
web.process_backend._use_pm2 = original_use_pm2
|
|
|
|
youtube_row = next(row for row in payload["entries"] if row["pm2"] == "youtube")
|
|
self.assertEqual(youtube_row["process_status"], "configured")
|
|
self.assertEqual(youtube_row["business_status"], "ready")
|
|
self.assertFalse(youtube_row["is_pushing"])
|
|
|
|
def test_process_monitor_family_filter_returns_youtube_rows_only(self) -> None:
|
|
original_use_pm2 = web.process_backend._use_pm2
|
|
web.process_backend._use_pm2 = False
|
|
try:
|
|
with patch("web.process_monitor_entries", return_value=(
|
|
{"pm2": "youtube", "script": "youtube.py", "label": "YouTube"},
|
|
{"pm2": "youtube2__lane1", "script": "youtube.py", "label": "Lane 1"},
|
|
{"pm2": "tiktok", "script": "tiktok.py", "label": "TikTok"},
|
|
)):
|
|
with patch.object(web.process_backend, "ensure_mode", AsyncMock()):
|
|
with patch.object(
|
|
web.process_backend,
|
|
"process_infos",
|
|
AsyncMock(
|
|
return_value={
|
|
"youtube": {"process_status": "stopped", "out_path": None, "err_path": None},
|
|
"youtube2__lane1": {"process_status": "online", "out_path": None, "err_path": None},
|
|
}
|
|
),
|
|
):
|
|
with patch("web.read_log_path_normalized", AsyncMock(return_value="")):
|
|
with patch.object(web.process_backend, "live_ffmpeg_pids", AsyncMock(return_value=[])):
|
|
payload = asyncio.run(web.process_monitor(family="youtube", light=True))
|
|
finally:
|
|
web.process_backend._use_pm2 = original_use_pm2
|
|
|
|
pm2_names = {row["pm2"] for row in payload["entries"]}
|
|
self.assertIn("youtube", pm2_names)
|
|
self.assertIn("youtube2__lane1", pm2_names)
|
|
self.assertTrue(all(name.startswith("youtube") or name.startswith("douyin_youtube") for name in pm2_names))
|
|
|
|
def test_restart_kills_old_ffmpeg_before_backend_restart(self) -> None:
|
|
calls: list[str] = []
|
|
|
|
async def fake_kill(process: str, base_dir, **_kwargs) -> None: # noqa: ANN001
|
|
calls.append(f"kill:{process}")
|
|
|
|
async def fake_restart(process: str, script: str) -> str:
|
|
calls.append(f"restart:{process}:{script}")
|
|
return "ok"
|
|
|
|
with patch("web.is_valid_control_process", return_value=True):
|
|
with patch("web.script_for_pm2_or_pro", return_value="youtube.py"):
|
|
with patch("web._validate_youtube_runtime_files", return_value=[]):
|
|
with patch("web._stop_other_hdmi_sink_processes", new=AsyncMock(return_value=[])):
|
|
with patch("web.force_kill_ffmpeg", new=AsyncMock(side_effect=fake_kill)):
|
|
with patch.object(web.process_backend, "restart", new=AsyncMock(side_effect=fake_restart)):
|
|
response = asyncio.run(web.restart("youtube"))
|
|
|
|
payload = json.loads(response.body)
|
|
self.assertEqual(payload["output"], "ok")
|
|
self.assertEqual(calls, ["kill:youtube", "restart:youtube:youtube.py"])
|
|
|
|
def test_get_url_config_does_not_resync_existing_managed_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config_dir = Path(tmp)
|
|
rel = "URL_config.youtube.ini"
|
|
(config_dir / rel).write_text("https://live.douyin.com/new\n", encoding="utf-8")
|
|
|
|
with patch("web.CONFIG_DIR", config_dir):
|
|
with patch("web.allows_url_config", return_value=True):
|
|
with patch("web.managed_url_config_relpath", return_value=rel):
|
|
with patch("web._sync_youtube_pro_runtime_files", side_effect=AssertionError("unexpected resync")):
|
|
payload = asyncio.run(web.get_url_config("youtube"))
|
|
|
|
self.assertEqual(_json_body(payload)["content"], "https://live.douyin.com/new\n")
|
|
self.assertEqual(payload.headers.get("cache-control"), "no-store, max-age=0, must-revalidate")
|
|
|
|
def test_save_url_config_updates_matching_youtube_pro_channel_store(self) -> None:
|
|
captured: list[list[dict[str, object]]] = []
|
|
channels = [{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube", "urlLines": "old\n"}]
|
|
|
|
with patch("web.allows_url_config", return_value=True):
|
|
with patch("web.managed_url_config_relpath", return_value="URL_config.youtube.ini"):
|
|
with patch("web.governed_save_managed_file"):
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
with patch("web.set_youtube_pro_channels", side_effect=lambda rows: captured.append(rows)):
|
|
payload = asyncio.run(
|
|
web.save_url_config(
|
|
_JsonRequest({"content": "https://live.douyin.com/fresh\n"}),
|
|
process="youtube",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(_json_body(payload)["message"], "配置保存成功")
|
|
self.assertEqual(captured[-1][0]["urlLines"], "https://live.douyin.com/fresh\n")
|
|
|
|
def test_save_config_updates_matching_youtube_pro_channel_stream_key(self) -> None:
|
|
captured: list[list[dict[str, object]]] = []
|
|
channels = [{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube", "streamKey": "old-key"}]
|
|
content = "[youtube]\nkey = new-key-123\nrtmps = 是\n"
|
|
|
|
with patch("web.managed_youtube_ini_relpath", return_value="youtube.youtube.ini"):
|
|
with patch("web.governed_save_managed_file"):
|
|
with patch("web.get_youtube_pro_channels", return_value=channels):
|
|
with patch("web.set_youtube_pro_channels", side_effect=lambda rows: captured.append(rows)):
|
|
payload = asyncio.run(
|
|
web.save_config(
|
|
_JsonRequest({"content": content}),
|
|
process="youtube",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(_json_body(payload)["message"], "配置保存成功")
|
|
self.assertEqual(captured[-1][0]["streamKey"], "new-key-123")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|