eric/gitlab-instance-0a899031_sh
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

s

Browse Source
This commit is contained in:
eric
2026-05-19 09:19:52 +00:00
parent 585bcd9500
commit b617555360
30 changed files with 477 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
d2ypp2/tests/test_web_dynamic_processes.py
View File

@@ -43,17 +43,19 @@ class WebDynamicProcessesTests(unittest.TestCase):
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")
resolved = web.script_for_pm2_or_pro("youtube2__custom-lane-three")
self.assertEqual(resolved, "youtube.py")
def test_process_monitor_entries_keep_static_youtube_label_for_lane_one(self) -> None:
def test_process_monitor_entries_migrates_legacy_youtube_pm2_without_hiding_single_lane(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")
pro_row = next(row for row in entries if row["pm2"] == "youtube2__lane1")
self.assertEqual(pro_row["label"], "YouTube Pro · Lane 1")
def test_process_monitor_entries_do_not_duplicate_static_pm2_rows(self) -> None:
channels = [
@@ -64,6 +66,7 @@ class WebDynamicProcessesTests(unittest.TestCase):
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__lane1"), 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:
@@ -183,6 +186,44 @@ class WebDynamicProcessesTests(unittest.TestCase):
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": "youtube2__lane1", "urlLines": "old\n"}]
with patch("web.allows_url_config", return_value=True):
with patch("web.managed_url_config_relpath", return_value="URL_config.youtube2__lane1.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="youtube2__lane1",
)
)
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": "youtube2__lane1", "streamKey": "old-key"}]
content = "[youtube]\nkey = new-key-123\nrtmps = 是\n"
with patch("web.managed_youtube_ini_relpath", return_value="youtube.youtube2__lane1.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="youtube2__lane1",
)
)
self.assertEqual(_json_body(payload)["message"], "配置保存成功")
self.assertEqual(captured[-1][0]["streamKey"], "new-key-123")
def test_single_youtube_save_does_not_mutate_legacy_pro_channel_named_youtube(self) -> None:
captured: list[list[dict[str, object]]] = []
channels = [{"id": "lane1", "name": "Lane 1", "pm2Name": "youtube", "urlLines": "old\n"}]
@@ -193,32 +234,13 @@ class WebDynamicProcessesTests(unittest.TestCase):
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"}),
_JsonRequest({"content": "https://live.douyin.com/single\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")
self.assertEqual(captured, [])
if __name__ == "__main__":