97 lines
4.5 KiB
Python
97 lines
4.5 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from douyin_youtube_ffplay import _load_youtube_push_settings, _resolve_youtube_push_config_file
|
|
from src.youtube_ini_sync import write_youtube_ini_stream_key
|
|
|
|
|
|
class YoutubePushConfigTests(unittest.TestCase):
|
|
def test_resolve_youtube_push_config_prefers_managed_pm2_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
cfg_dir = root / "config"
|
|
cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
(cfg_dir / "youtube.ini").write_text("[youtube]\nkey = legacy\n", encoding="utf-8")
|
|
managed = cfg_dir / "youtube.youtube.ini"
|
|
managed.write_text("[youtube]\nkey = managed\n", encoding="utf-8")
|
|
|
|
with patch.dict("os.environ", {"LIVE_PM2_PROCESS_NAME": "youtube"}, clear=False):
|
|
resolved = _resolve_youtube_push_config_file(base_dir=str(root))
|
|
|
|
self.assertEqual(resolved, str(managed))
|
|
|
|
def test_resolve_youtube_push_config_falls_back_to_legacy_file(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
cfg_dir = root / "config"
|
|
cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
legacy = cfg_dir / "youtube.ini"
|
|
legacy.write_text("[youtube]\nkey = legacy\n", encoding="utf-8")
|
|
|
|
with patch.dict("os.environ", {"LIVE_PM2_PROCESS_NAME": "youtube"}, clear=False):
|
|
resolved = _resolve_youtube_push_config_file(base_dir=str(root))
|
|
|
|
self.assertEqual(resolved, str(legacy))
|
|
|
|
def test_write_youtube_ini_stream_key_updates_default_managed_file_when_present(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
cfg_dir = root / "config"
|
|
cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
(cfg_dir / "youtube.ini").write_text("[youtube]\nkey = old\n", encoding="utf-8")
|
|
(cfg_dir / "youtube.youtube.ini").write_text("[youtube]\nkey = old2\n", encoding="utf-8")
|
|
|
|
writes: list[tuple[str, str, str]] = []
|
|
|
|
def fake_save(root_id: str, relative_path: str, content: str) -> dict[str, str]:
|
|
writes.append((root_id, relative_path, content))
|
|
return {"root_id": root_id, "path": relative_path, "message": "Saved successfully"}
|
|
|
|
with patch("src.youtube_ini_sync.save_managed_file", side_effect=fake_save):
|
|
msg = write_youtube_ini_stream_key(root, "demo-key")
|
|
|
|
self.assertIn("youtube.ini", msg)
|
|
self.assertIn("youtube.youtube.ini", msg)
|
|
self.assertEqual([w[1] for w in writes], ["youtube.ini", "youtube.youtube.ini"])
|
|
self.assertTrue(all("demo-key" in w[2] for w in writes))
|
|
|
|
def test_load_youtube_push_settings_accepts_bom_prefixed_ini(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
cfg = Path(tmp) / "youtube.youtube.ini"
|
|
cfg.write_text("\ufeff[youtube]\nkey = bom-key\nbitrate = 5500\nfast_audio = 1\n", encoding="utf-8")
|
|
|
|
parsed = _load_youtube_push_settings(str(cfg))
|
|
|
|
self.assertEqual(parsed["stream_key"], "bom-key")
|
|
self.assertEqual(parsed["b_v"], "5500k")
|
|
self.assertEqual(parsed["maxrate"], "6600k")
|
|
self.assertEqual(parsed["bufsize"], "11000k")
|
|
self.assertTrue(parsed["fast_audio"])
|
|
|
|
def test_write_youtube_ini_stream_key_reads_existing_bom_prefixed_ini(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
root = Path(tmp)
|
|
cfg_dir = root / "config"
|
|
cfg_dir.mkdir(parents=True, exist_ok=True)
|
|
(cfg_dir / "youtube.youtube.ini").write_text("\ufeff[youtube]\nkey = old\n", encoding="utf-8")
|
|
|
|
writes: list[tuple[str, str, str]] = []
|
|
|
|
def fake_save(root_id: str, relative_path: str, content: str) -> dict[str, str]:
|
|
writes.append((root_id, relative_path, content))
|
|
return {"root_id": root_id, "path": relative_path, "message": "Saved successfully"}
|
|
|
|
with patch("src.youtube_ini_sync.save_managed_file", side_effect=fake_save):
|
|
msg = write_youtube_ini_stream_key(root, "fresh-key")
|
|
|
|
self.assertIn("youtube.youtube.ini", msg)
|
|
self.assertTrue(any(path == "youtube.youtube.ini" and "fresh-key" in content for _, path, content in writes))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|