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()