vmess-domain-rotator/tests/test_server_mode_lifecycle.py

import json
import os
import pathlib
import shutil
import subprocess
import sys
import tempfile
import unittest

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


class TestServerModeFullLifecycle(unittest.TestCase):
    """Server 模式（除安装与卸载外）全流程本地端到端测试套件"""

    def setUp(self):
        self.tmp_dir = tempfile.TemporaryDirectory()
        self.work_dir = pathlib.Path(self.tmp_dir.name)

    def tearDown(self):
        self.tmp_dir.cleanup()

    def _setup_mock_repo(self) -> pathlib.Path:
        """初始化一个真实的隔离 Git 仓库并克隆项目脚本与配置"""
        repo_dir = self.work_dir / "target_repo"
        repo_dir.mkdir()
        subprocess.run(["git", "init", "-b", "main"], cwd=str(repo_dir), check=True, capture_output=True)
        subprocess.run(["git", "config", "user.name", "Tester"], cwd=str(repo_dir), check=True)
        subprocess.run(["git", "config", "user.email", "tester@example.com"], cwd=str(repo_dir), check=True)

        # 拷贝必要目录与脚本
        shutil.copytree(PROJECT_ROOT / "scripts", repo_dir / "scripts")
        shutil.copytree(PROJECT_ROOT / "configs", repo_dir / "configs")

        # 提交初始版本以确立 HEAD
        subprocess.run(["git", "add", "."], cwd=str(repo_dir), check=True)
        subprocess.run(["git", "commit", "-m", "initial commit"], cwd=str(repo_dir), check=True, capture_output=True)

        return repo_dir

    def _write_mock_api_payload(self, records: list) -> pathlib.Path:
        """写入符合 vps789.com OpenAPI 格式的本地 JSON 文件"""
        payload_file = self.work_dir / "vps789_api_response.json"
        payload = {
            "code": 200,
            "message": "success",
            "data": records,
        }
        payload_file.write_text(json.dumps(payload, ensure_ascii=False, indent=2))
        return payload_file

    def test_server_mode_full_lifecycle_api_fetch_and_git_commit(self):
        """[全流程 1] API 拉取 -> 候选过滤 -> 加权选优 -> 运行时落盘 -> Worktree 隔离分支提交"""
        repo_dir = self._setup_mock_repo()
        config_path = repo_dir / "configs" / "config.server.json"

        # 构造典型候选数据集
        records = [
            # 候选 1: 优质节点 (移动延迟 38ms, 丢包 0, 健康分 105) -> 加权得分 38.0，应胜出
            {
                "ip": "best.node.example.com",
                "createdTime": "2026-06-01 12:00:00",
                "avgLatency": "80",
                "avgPkgLostRate": "0",
                "ydLatency": 38.0,
                "ydPkgLostRate": 0.0,
                "ydScore": 105.0,
                "hostProvider": "Official",
            },
            # 候选 2: 次优节点 (移动延迟 65ms, 丢包 0, 健康分 90) -> 加权得分 65.0
            {
                "ip": "second.node.example.com",
                "createdTime": "2026-06-01 12:00:00",
                "avgLatency": "95",
                "avgPkgLostRate": "0",
                "ydLatency": 65.0,
                "ydPkgLostRate": 0.0,
                "ydScore": 90.0,
                "hostProvider": "Official",
            },
            # 候选 3: 黑名单提供商，必须被 record_filter.exclude_if_any 过滤
            {
                "ip": "filtered.cf.example.com",
                "createdTime": "2026-06-01 12:00:00",
                "avgLatency": "20",
                "avgPkgLostRate": "0",
                "ydLatency": 20.0,
                "ydPkgLostRate": 0.0,
                "ydScore": 110.0,
                "hostProvider": "CF优选IP",
            },
            # 候选 4: 纯 IP，必须被 domain_filter 过滤
            {
                "ip": "198.51.100.1",
                "createdTime": "2026-06-01 12:00:00",
                "avgLatency": "10",
                "avgPkgLostRate": "0",
                "ydLatency": 10.0,
                "ydPkgLostRate": 0.0,
                "ydScore": 110.0,
                "hostProvider": "Official",
            },
        ]
        payload_file = self._write_mock_api_payload(records)

        # 替换 config.server.json 中的 URL 为 file:// 路径
        with open(config_path, "r", encoding="utf-8") as f:
            cfg = json.load(f)
        cfg["api"]["url"] = f"file://{payload_file}"
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(cfg, f, indent=2, ensure_ascii=False)

        script_path = repo_dir / "scripts" / "server" / "run_update_and_commit.sh"
        env = os.environ.copy()
        env["APP_DIR"] = str(repo_dir)
        env["GIT_RUNTIME_BRANCH"] = "runtime-state"
        env["GIT_PUSH_ENABLED"] = "0"
        env["PYTHON_BIN"] = sys.executable

        # 首次执行：通过 --force-commit 触发完整工作流
        proc = subprocess.run(
            ["bash", str(script_path), "--force-commit", str(config_path)],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(repo_dir),
        )
        self.assertEqual(proc.returncode, 0, f"执行失败:\nSTDOUT: {proc.stdout}\nSTDERR: {proc.stderr}")

        # 1. 验证落盘文件正确性 (锚定在 repo_dir/runtime，决不是 repo_dir/configs/runtime)
        runtime_dir = repo_dir / "runtime"
        self.assertTrue(runtime_dir.exists(), "必须生成 runtime 目录")
        self.assertFalse((repo_dir / "configs" / "runtime").exists(), "严禁在 configs/ 下生成 runtime")

        txt_file = runtime_dir / "current_domain.txt"
        json_file = runtime_dir / "current_domain.json"
        state_file = runtime_dir / "state.json"
        vars_file = runtime_dir / "substore_vars.json"

        self.assertTrue(txt_file.exists())
        self.assertTrue(json_file.exists())
        self.assertTrue(state_file.exists())
        self.assertTrue(vars_file.exists())

        # 2. 验证选优结果（best.node.example.com 胜出）
        selected_domain = txt_file.read_text().strip()
        self.assertEqual(selected_domain, "best.node.example.com")

        # 3. 验证 JSON 结构和 Sub-Store 契约
        json_data = json.loads(json_file.read_text())
        self.assertEqual(json_data["domain"], "best.node.example.com")
        self.assertEqual(json_data["status"], "ok")
        self.assertIn("top_candidates", json_data)
        # 验证过滤规则生效：CF优选IP 与纯 IP 不在候选列表中
        candidate_domains = [c["domain"] for c in json_data["top_candidates"]]
        self.assertNotIn("filtered.cf.example.com", candidate_domains)
        self.assertNotIn("198.51.100.1", candidate_domains)

        vars_data = json.loads(vars_file.read_text())
        self.assertEqual(vars_data["AUTO_DOMAIN"], "best.node.example.com")
        self.assertEqual(vars_data["STATUS"], "ok")

        # 4. 验证 Git 分支已自动建立孤儿分支 runtime-state 并写入 commit
        commit_log = subprocess.check_output(
            ["git", "log", "-n", "1", "--format=%B", "runtime-state"],
            cwd=str(repo_dir),
            text=True,
        ).strip()
        self.assertIn("manual: value best.node.example.com", commit_log)

        # 5. 验证孤儿分支上的文件确实位于 runtime/ 路径下
        tree_files = subprocess.check_output(
            ["git", "ls-tree", "-r", "--name-only", "runtime-state"],
            cwd=str(repo_dir),
            text=True,
        ).splitlines()
        self.assertIn("runtime/current_domain.txt", tree_files)
        self.assertIn("runtime/current_domain.json", tree_files)
        self.assertIn("runtime/substore_vars.json", tree_files)
        self.assertIn("runtime/state.json", tree_files)

    def test_server_mode_adaptive_interval_window(self):
        """[全流程 2] 动态自适应执行窗口：非高峰期非整点窗口跳过执行，高峰期窗口正常推进"""
        repo_dir = self._setup_mock_repo()
        config_path = repo_dir / "configs" / "config.server.json"
        script_path = repo_dir / "scripts" / "server" / "run_update_and_commit.sh"

        env = os.environ.copy()
        env["APP_DIR"] = str(repo_dir)
        env["GIT_RUNTIME_BRANCH"] = "runtime-state"
        env["GIT_PUSH_ENABLED"] = "0"
        env["PYTHON_BIN"] = sys.executable

        # 场景 A: 设定高峰期为 02:00 - 03:00，周期 60 分钟，容差 1 分钟
        # 若当前时间不是 02 点且不是整点，必定触发 off-peak run skipped
        env_skip = env.copy()
        env_skip["PEAK_START_HOUR"] = "2"
        env_skip["PEAK_END_HOUR"] = "3"
        env_skip["OFFPEAK_INTERVAL_MIN"] = "60"
        env_skip["PEAK_INTERVAL_MIN"] = "2"

        proc_skip = subprocess.run(
            ["bash", str(script_path), str(config_path)],
            env=env_skip,
            capture_output=True,
            text=True,
            cwd=str(repo_dir),
        )
        self.assertEqual(proc_skip.returncode, 0)
        # 如果恰逢整点则可能执行，否则必定跳过；测试两种合法分支
        combined_skip = f"{proc_skip.stdout}\n{proc_skip.stderr}"
        if "off-peak run skipped" in combined_skip:
            self.assertIn("boundary", combined_skip)

        # 场景 B: 设定全天 00:00 - 24:00 均为高峰期，必定不跳过自适应窗口
        payload_file = self._write_mock_api_payload([
            {"ip": "peak.node.example.com", "ydLatency": 30.0, "ydPkgLostRate": 0.0, "ydScore": 100.0}
        ])
        with open(config_path, "r", encoding="utf-8") as f:
            cfg = json.load(f)
        cfg["api"]["url"] = f"file://{payload_file}"
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(cfg, f, indent=2, ensure_ascii=False)

        env_peak = env.copy()
        env_peak["PEAK_START_HOUR"] = "0"
        env_peak["PEAK_END_HOUR"] = "24"
        proc_peak = subprocess.run(
            ["bash", str(script_path), str(config_path)],
            env=env_peak,
            capture_output=True,
            text=True,
            cwd=str(repo_dir),
        )
        self.assertEqual(proc_peak.returncode, 0)
        combined_peak = f"{proc_peak.stdout}\n{proc_peak.stderr}"
        self.assertNotIn("off-peak run skipped", combined_peak, "高峰期窗口内绝不应跳过")
        self.assertIn("peak.node.example.com", (repo_dir / "runtime" / "current_domain.txt").read_text())

    def test_server_mode_fallback_and_silent_skip(self):
        """[全流程 3] 上游故障回退与静默跳过：API 故障回退至 last_good，且未变化时静默跳过 commit"""
        repo_dir = self._setup_mock_repo()
        config_path = repo_dir / "configs" / "config.server.json"

        # 指向一个不存在的文件 URL 模拟 API 失败
        with open(config_path, "r", encoding="utf-8") as f:
            cfg = json.load(f)
        cfg["api"]["url"] = "file:///nonexistent/path/never_found.json"
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(cfg, f, indent=2, ensure_ascii=False)

        # 预先在 state.json 中写入 last_good_domain
        runtime_dir = repo_dir / "runtime"
        runtime_dir.mkdir(parents=True, exist_ok=True)
        (runtime_dir / "state.json").write_text(json.dumps({
            "last_good_domain": "fallback-anchor.example.com",
            "last_run": "2026-06-01T00:00:00Z"
        }))

        script_path = repo_dir / "scripts" / "server" / "run_update_and_commit.sh"
        env = os.environ.copy()
        env["APP_DIR"] = str(repo_dir)
        env["GIT_RUNTIME_BRANCH"] = "runtime-state"
        env["GIT_PUSH_ENABLED"] = "0"
        env["PYTHON_BIN"] = sys.executable
        # 强制设置全天为高峰，防止被自适应时间窗口拦截
        env["PEAK_START_HOUR"] = "0"
        env["PEAK_END_HOUR"] = "24"

        # 1. 首次运行（强制提交），建立 fallback commit
        proc1 = subprocess.run(
            ["bash", str(script_path), "--force-commit", str(config_path)],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(repo_dir),
        )
        self.assertEqual(proc1.returncode, 0)
        self.assertIn("warning: updater using fallback value", f"{proc1.stdout}\n{proc1.stderr}")
        self.assertEqual((runtime_dir / "current_domain.txt").read_text().strip(), "fallback-anchor.example.com")

        # 2. 第二次运行（普通运行，无 --force-commit）：由于仍处于 fallback 状态且值未变化，应执行静默跳过！
        proc2 = subprocess.run(
            ["bash", str(script_path), str(config_path)],
            env=env,
            capture_output=True,
            text=True,
            cwd=str(repo_dir),
        )
        combined2 = f"{proc2.stdout}\n{proc2.stderr}"
        self.assertEqual(proc2.returncode, 0)
        self.assertIn("selected value unchanged (fallback-anchor.example.com)", combined2)
        self.assertIn("skip git commit and push", combined2)

    def test_server_mode_rolling_history_prune(self):
        """[全流程 4] 历史修剪 (Mode A): 孤儿分支提交数超过 GIT_MAX_STATE_COMMITS 时自动 Squashed Rebase"""
        repo_dir = self._setup_mock_repo()
        config_path = repo_dir / "configs" / "config.server.json"
        script_path = repo_dir / "scripts" / "server" / "run_update_and_commit.sh"

        env = os.environ.copy()
        env["APP_DIR"] = str(repo_dir)
        env["GIT_RUNTIME_BRANCH"] = "runtime-state"
        env["GIT_PUSH_ENABLED"] = "0"
        env["GIT_MAX_STATE_COMMITS"] = "3"  # 最大只保留 3 个 commit
        env["PYTHON_BIN"] = sys.executable

        # 循环 5 次变换最优节点，促使产生 5 次不同的 commit
        for i in range(1, 6):
            payload_file = self._write_mock_api_payload([
                {
                    "ip": f"node-{i}.example.com",
                    "createdTime": "2026-06-01 12:00:00",
                    "avgLatency": "50",
                    "avgPkgLostRate": "0",
                    "ydLatency": 20.0 + i,
                    "ydPkgLostRate": 0.0,
                    "ydScore": 100.0,
                    "hostProvider": "Official",
                }
            ])
            with open(config_path, "r", encoding="utf-8") as f:
                cfg = json.load(f)
            cfg["api"]["url"] = f"file://{payload_file}"
            with open(config_path, "w", encoding="utf-8") as f:
                json.dump(cfg, f, indent=2, ensure_ascii=False)

            proc = subprocess.run(
                ["bash", str(script_path), "--force-commit", str(config_path)],
                env=env,
                capture_output=True,
                text=True,
                cwd=str(repo_dir),
            )
            self.assertEqual(proc.returncode, 0)

        # 检查 runtime-state 的 commit 总数：必须严格等于 GIT_MAX_STATE_COMMITS (3)
        final_count = int(subprocess.check_output(
            ["git", "rev-list", "--count", "runtime-state"],
            cwd=str(repo_dir),
            text=True,
        ).strip())
        self.assertEqual(final_count, 3, "超过阈值后必须自动修剪至 3 个 commit")


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