138 lines
6.0 KiB
Python
138 lines
6.0 KiB
Python
from __future__ import annotations
|
|
|
|
import errno
|
|
import shutil
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from src.config_governance import (
|
|
analyze_content,
|
|
backup_managed_file,
|
|
optimize_content,
|
|
save_managed_file,
|
|
)
|
|
from src.control_plane import ManagedRoot
|
|
|
|
|
|
class ConfigGovernanceTests(unittest.TestCase):
|
|
def test_url_config_detects_duplicates(self) -> None:
|
|
sample = "\n".join(
|
|
[
|
|
"https://live.douyin.com/123456,主播A",
|
|
"https://live.douyin.com/123456,主播A重复",
|
|
"# https://live.douyin.com/999999",
|
|
"https://www.tiktok.com/@demo/live",
|
|
]
|
|
)
|
|
result = analyze_content("app-config", "URL_config.ini", sample, source="draft")
|
|
codes = {item["code"] for item in result["issues"]}
|
|
self.assertIn("duplicate_urls", codes)
|
|
self.assertTrue(any(item["id"] == "dedupe_urls" for item in result["actions"]))
|
|
|
|
def test_dedupe_urls_keeps_single_active_copy(self) -> None:
|
|
sample = "\n".join(
|
|
[
|
|
"# keep comment",
|
|
"https://live.douyin.com/123456",
|
|
"https://live.douyin.com/123456",
|
|
"https://live.douyin.com/654321",
|
|
]
|
|
)
|
|
result = optimize_content("app-config", "URL_config.ini", "dedupe_urls", sample)
|
|
content = result["content"]
|
|
self.assertEqual(content.count("https://live.douyin.com/123456"), 1)
|
|
self.assertIn("# keep comment", content)
|
|
self.assertIn("https://live.douyin.com/654321", content)
|
|
|
|
def test_youtube_preset_preserves_key_and_enables_balanced_defaults(self) -> None:
|
|
sample = "[youtube]\nkey = demo-key\nrtmps = 否\nfast_audio = 0\n"
|
|
result = optimize_content("app-config", "youtube.ini", "youtube_balanced_preset", sample)
|
|
content = result["content"]
|
|
self.assertIn("key = demo-key", content)
|
|
self.assertIn("rtmps = 是", content)
|
|
self.assertIn("bitrate = 4500", content)
|
|
self.assertIn("fast_audio = 1", content)
|
|
|
|
def test_backup_managed_file_reclaims_history_when_disk_full(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
root_path = tmp_path / "config"
|
|
root_path.mkdir(parents=True, exist_ok=True)
|
|
target = root_path / "youtube.demo.ini"
|
|
target.write_text("[youtube]\nkey = old\n", encoding="utf-8")
|
|
root = ManagedRoot(
|
|
root_id="app-config",
|
|
label="App Config",
|
|
description="test",
|
|
path=root_path,
|
|
explicit_names=("youtube.demo.ini",),
|
|
)
|
|
hist_dir = tmp_path / "backup_config" / "__web__" / "app-config" / "youtube.demo.ini.history"
|
|
hist_dir.mkdir(parents=True, exist_ok=True)
|
|
(hist_dir / "stale-1.ini").write_text("1", encoding="utf-8")
|
|
(hist_dir / "stale-2.ini").write_text("2", encoding="utf-8")
|
|
|
|
calls = {"n": 0}
|
|
real_copy2 = shutil.copy2
|
|
|
|
def flaky_copy2(src, dst, *args, **kwargs):
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise OSError(errno.ENOSPC, "No space left on device")
|
|
return real_copy2(src, dst, *args, **kwargs)
|
|
|
|
with mock.patch("src.config_governance.resolve_managed_file", return_value=(root, Path("youtube.demo.ini"), target)):
|
|
with mock.patch("src.config_governance.WEB_BACKUP_DIR", tmp_path / "backup_config" / "__web__"):
|
|
with mock.patch("src.config_governance.shutil.copy2", side_effect=flaky_copy2):
|
|
backup = backup_managed_file("app-config", "youtube.demo.ini", reason="save")
|
|
|
|
self.assertIsNotNone(backup)
|
|
self.assertGreaterEqual(calls["n"], 2)
|
|
remaining = sorted(p.name for p in hist_dir.iterdir() if p.is_file())
|
|
self.assertEqual(len(remaining), 1)
|
|
self.assertTrue(remaining[0].endswith(".ini"))
|
|
|
|
def test_save_managed_file_retries_after_enospc(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
tmp_path = Path(tmp)
|
|
root_path = tmp_path / "config"
|
|
root_path.mkdir(parents=True, exist_ok=True)
|
|
target = root_path / "youtube.demo.ini"
|
|
target.write_text("[youtube]\nkey = old\n", encoding="utf-8")
|
|
root = ManagedRoot(
|
|
root_id="app-config",
|
|
label="App Config",
|
|
description="test",
|
|
path=root_path,
|
|
explicit_names=("youtube.demo.ini",),
|
|
)
|
|
hist_dir = tmp_path / "backup_config" / "__web__" / "app-config" / "youtube.demo.ini.history"
|
|
hist_dir.mkdir(parents=True, exist_ok=True)
|
|
(hist_dir / "stale.ini").write_text("stale", encoding="utf-8")
|
|
|
|
calls = {"n": 0}
|
|
real_write_text = Path.write_text
|
|
|
|
def flaky_write_text(self, *args, **kwargs):
|
|
if self == target:
|
|
calls["n"] += 1
|
|
if calls["n"] == 1:
|
|
raise OSError(errno.ENOSPC, "No space left on device")
|
|
return real_write_text(self, *args, **kwargs)
|
|
|
|
with mock.patch("src.config_governance.resolve_managed_file", return_value=(root, Path("youtube.demo.ini"), target)):
|
|
with mock.patch("src.config_governance.WEB_BACKUP_DIR", tmp_path / "backup_config" / "__web__"):
|
|
with mock.patch("pathlib.Path.write_text", new=flaky_write_text):
|
|
result = save_managed_file("app-config", "youtube.demo.ini", "[youtube]\nkey = new\n")
|
|
|
|
self.assertEqual(result["message"], "Saved successfully")
|
|
self.assertEqual(target.read_text(encoding="utf-8"), "[youtube]\nkey = new\n")
|
|
self.assertGreaterEqual(calls["n"], 2)
|
|
self.assertFalse(hist_dir.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|