vmess-domain-rotator/tests/test_router_scripts.py

import pathlib
import unittest

PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent


class TestRouterScriptsBugsAndContracts(unittest.TestCase):
    """测试并锁定路由器 BusyBox 脚本中的文件原子性与选优漏洞，并纠正旧报告虚假报警"""

    def test_p1_router_local_update_atomic_file_write(self):
        """[P1/P2 修复验证] 验证 router_local_update.sh 采用 .tmp.$$ + mv -f 原子替换写入文件"""
        script_path = PROJECT_ROOT / "scripts" / "router" / "router_local_update.sh"
        content = script_path.read_text()

        self.assertIn("atomic_write_file()", content)
        self.assertIn(".tmp.$$", content)
        self.assertIn('atomic_write_file "$CURRENT_JSON_PATH"', content)

    def test_p2_router_local_update_filters_zero_speed(self):
        """[P2 修复验证] 验证 router_local_update.sh 在开启测速时优先挑选 download_speed > 0 的候选行"""
        script_path = PROJECT_ROOT / "scripts" / "router" / "router_local_update.sh"
        content = script_path.read_text()

        self.assertIn("speed = trim($6) + 0", content)
        self.assertIn("speed > 0", content)

    def test_recheck_correction_router_cfst_params_already_supported(self):
        """[复核纠偏] 验证初审报告关于 'router_local_update.sh 缺少 -dt/-dn/-dd 参数' 系虚假误报"""
        script_path = PROJECT_ROOT / "scripts" / "router" / "router_local_update.sh"
        content = script_path.read_text()

        # 验证源码中实际已存在这些参数映射
        self.assertIn('CFST_DOWNLOAD_COUNT', content)
        self.assertIn('-dn "$CFST_DOWNLOAD_COUNT"', content)
        self.assertIn('CFST_DOWNLOAD_TIME', content)
        self.assertIn('-dt "$CFST_DOWNLOAD_TIME"', content)
        self.assertIn('CFST_DISABLE_DOWNLOAD', content)
        self.assertIn('-dd', content)

    def test_build_router_package_ssot_sync(self):
        """[方案B SSOT驱动验证] 验证 build_router_package.sh --sync-from-config 自动将 JSON 测速与输出参数投影至路由器配置"""
        import json
        import os
        import subprocess
        import tarfile
        import tempfile

        with tempfile.TemporaryDirectory() as tmpdir:
            tmppath = pathlib.Path(tmpdir)
            base_conf = tmppath / "base_router.conf"
            base_conf.write_text("""CFST_DISPLAY_COUNT="10"
CFST_PORT="443"
CFST_DISABLE_DOWNLOAD="0"
TOP_N="3"
VALUE_TEXT_FILE="current_ip.txt"
""")

            json_conf = tmppath / "config.local.json"
            json_conf.write_text(json.dumps({
                "selection": {"top_n": 5},
                "output": {"selected_value_file": "ssot_custom_ip.txt"},
                "cfst_local": {
                    "run_args": ["-p", "25", "-f", "my_ips.txt", "-tp", "8443", "-dd"]
                }
            }))

            out_tar = tmppath / "test_pkg.tar.gz"
            script_path = PROJECT_ROOT / "build_router_package.sh"

            env = os.environ.copy()
            env["ROUTER_CONF"] = str(base_conf)
            env["OUT_TAR"] = str(out_tar)
            env["TMP_DIR"] = str(tmppath / "build_tmp")

            proc = subprocess.run(
                [
                    "sh",
                    str(script_path),
                    "--sync-from-config",
                    str(json_conf),
                ],
                env=env,
                capture_output=True,
                text=True,
                cwd=str(PROJECT_ROOT),
            )

            self.assertEqual(proc.returncode, 0, f"打包失败: {proc.stderr}\nSTDOUT: {proc.stdout}")
            self.assertTrue(out_tar.exists(), "生成的 tar.gz 必须存在")

            # 解压并校验打包中的 config_router.conf
            with tarfile.open(out_tar, "r:gz") as tar:
                conf_member = tar.extractfile("vmess/config_router.conf")
                self.assertIsNotNone(conf_member)
                conf_content = conf_member.read().decode("utf-8")

            # 验证 SSOT 同步生效
            self.assertIn('TOP_N="5"', conf_content)
            self.assertIn('VALUE_TEXT_FILE="ssot_custom_ip.txt"', conf_content)
            self.assertIn('CFST_DISPLAY_COUNT="25"', conf_content)
            self.assertIn('CFST_IP_FILE="my_ips.txt"', conf_content)
            self.assertIn('CFST_PORT="8443"', conf_content)
            self.assertIn('CFST_DISABLE_DOWNLOAD="1"', conf_content)


if __name__ == "__main__":
    unittest.main()