vmess-domain-rotator/tests/test_domain_updater.py

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

# 将 scripts 目录加入 sys.path
PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent
sys.path.insert(0, str(PROJECT_ROOT / "scripts"))

import domain_updater as du


class TestDomainUpdaterP0P1Bugs(unittest.TestCase):
    """锁定 domain_updater.py 与 config.server.json 中的 P0/P1 Bug"""

    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 test_p0_yd_score_polarity_fix_and_healthy_ranking(self):
        """[P0-1 修复验证] 验证 yd_score 极性修复与平局决胜：
        1. 加权公式仅计算 avg_latency (1.0) 与 loss_rate (10.0)
        2. 恶化前 fastsoso (36ms) 优于 competitor (49ms)
        3. 恶化后 fastsoso (74ms) 得分劣于 competitor (49ms)，competitor 成功登顶第 1
        4. 平局时 tie_breakers 优先选用 yd_score 降序（健康分高者胜出）
        """
        config_path = PROJECT_ROOT / "config.server.json"
        with open(config_path, "r", encoding="utf-8") as f:
            cfg = json.load(f)

        field_map = cfg["record_mapping"]["field_map"]
        record_mapping_cfg = cfg["record_mapping"]
        scoring_cfg = cfg["scoring"]

        # 真实数据：fastsoso 正常 vs 恶化
        normal_candidate = {
            "ip": "www.fastsoso.cn",
            "createdTime": "2026-05-30 23:42:59",
            "avgLatency": "94",
            "avgPkgLostRate": "0",
            "ydLatency": 36.0,
            "ydPkgLostRate": 0.0,
            "ydScore": 109.0,  # 健康分高
        }
        degraded_candidate = {
            "ip": "www.fastsoso.cn",
            "createdTime": "2026-05-30 23:42:59",
            "avgLatency": "115",
            "avgPkgLostRate": "0",
            "ydLatency": 74.0,  # 延迟翻倍恶化
            "ydPkgLostRate": 0.0,
            "ydScore": 78.0,   # 健康分暴跌
        }
        competitor = {
            "ip": "saas.sin.fan",
            "createdTime": "2026-06-08 21:51:52",
            "avgLatency": "98",
            "avgPkgLostRate": "0",
            "ydLatency": 49.0,
            "ydPkgLostRate": 0.0,
            "ydScore": 86.0,
        }

        # 1. 正常时算分：fastsoso 延迟更低，加权分更低更优，排名第一
        scored_normal = du.parse_scored_records([normal_candidate, competitor], field_map, record_mapping_cfg, scoring_cfg)
        ranked_normal = du.rank_scored_records(scored_normal, scoring_cfg)
        self.assertEqual(ranked_normal[0]["domain"], "www.fastsoso.cn")

        # 2. 恶化后算分：fastsoso 延迟翻倍恶化，competitor 成功夺得第 1 名
        scored_degraded = du.parse_scored_records([degraded_candidate, competitor], field_map, record_mapping_cfg, scoring_cfg)
        ranked_degraded = du.rank_scored_records(scored_degraded, scoring_cfg)
        self.assertEqual(ranked_degraded[0]["domain"], "saas.sin.fan", "恶化后 competitor 夺得第 1 名")
        self.assertEqual(ranked_degraded[1]["domain"], "www.fastsoso.cn", "恶化节点被挤至第 2 名")

        # 3. 平局决胜验证：延迟丢包相同，yd_score 高者胜出
        tie_candidate_a = {
            "ip": "tie-a.com",
            "createdTime": "2026-05-30 20:00:00",
            "avgLatency": "50",
            "avgPkgLostRate": "0",
            "ydLatency": 50.0,
            "ydPkgLostRate": 0.0,
            "ydScore": 99.0,
        }
        tie_candidate_b = {
            "ip": "tie-b.com",
            "createdTime": "2026-05-30 20:00:00",
            "avgLatency": "50",
            "avgPkgLostRate": "0",
            "ydLatency": 50.0,
            "ydPkgLostRate": 0.0,
            "ydScore": 70.0,
        }
        scored_tie = du.parse_scored_records([tie_candidate_b, tie_candidate_a], field_map, record_mapping_cfg, scoring_cfg)
        ranked_tie = du.rank_scored_records(scored_tie, scoring_cfg)
        self.assertEqual(ranked_tie[0]["domain"], "tie-a.com", "平局时 yd_score 降序胜出")

    def test_p0_notify_stdout_isolation_and_timeout(self):
        """[P0/P1 修复验证] 验证 run_notify() 标准输出重定向至 stderr，且具备超时保护"""
        cmd = [
            sys.executable,
            "-c",
            "import sys; sys.path.insert(0, 'scripts'); import domain_updater as du; du.run_notify('echo POLLUTING_STDOUT_LINE', 'test.com', 'ok')",
        ]
        proc = subprocess.run(cmd, capture_output=True, text=True, cwd=str(PROJECT_ROOT))
        self.assertNotIn("POLLUTING_STDOUT_LINE", proc.stdout, "run_notify 输出绝不泄露到 stdout")
        self.assertIn("POLLUTING_STDOUT_LINE", proc.stderr, "run_notify 输出已被安全重定向至 stderr")

        import unittest.mock as mock
        with mock.patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd="mock_notify", timeout=15)):
            with io.StringIO() as fake_stderr:
                with mock.patch("sys.stderr", fake_stderr):
                    du.run_notify("mock_timeout_cmd", "test.com", "ok")
                self.assertIn("timed out", fake_stderr.getvalue())

    def test_p1_error_fallback_stdout_contains_domain_key(self):
        """[P1 修复验证] 验证 domain_updater 异常回退输出的 JSON 包含配置的 domain/ip 键"""
        state_file = self.work_dir / "state.json"
        with open(state_file, "w", encoding="utf-8") as f:
            json.dump({"last_good": "fallback.example.com", "updated_at": "2026-01-01T00:00:00Z"}, f)

        cfg_copy = {
            "source": {"type": "api"},
            "api": {"url": "http://127.0.0.1:9/nonexistent", "timeout": 1},
            "record_mapping": {
                "records_path": "data[]",
                "field_map": {"domain": "ip", "created_at": "createdTime"},
                "created_time_formats": ["%Y-%m-%d %H:%M:%S"],
            },
            "output": {
                "runtime_dir": str(self.work_dir),
                "selected_value_file": "current.txt",
                "selected_value_json": "current.json",
                "selected_value_json_key": "domain",
                "state_file": "state.json",
                "state_last_good_key": "last_good",
                "export_vars_file": "vars.json",
                "substore_value_key": "DOMAIN",
            },
        }
        cfg_path = self.work_dir / "config.json"
        with open(cfg_path, "w", encoding="utf-8") as f:
            json.dump(cfg_copy, f)

        cmd = [
            sys.executable,
            str(PROJECT_ROOT / "scripts" / "domain_updater.py"),
            "--config", str(cfg_path),
        ]
        proc = subprocess.run(cmd, capture_output=True, text=True)
        self.assertEqual(proc.returncode, 0)
        lines = [line.strip() for line in proc.stdout.splitlines() if line.strip()]
        last_json = json.loads(lines[-1])
        self.assertEqual(last_json.get("status"), "error_use_last_good")
        self.assertEqual(last_json.get("domain"), "fallback.example.com", "异常回退输出成功包含 domain 键")

    def test_p1_last_good_bypasses_blacklist_prevented(self):
        """[P1 修复验证] 验证 last_good 命中黑名单时绝不借回退复活"""
        filter_cfg = {
            "include_suffixes": [],
            "exclude_regex": ["bad-domain\\.com"],
        }
        self.assertEqual(du.apply_filter(["bad-domain.com"], filter_cfg), [])

        # 主流程中若只有拉黑的 last_good，必须抛出异常而非选用
        last_good = "bad-domain.com"
        if du.apply_filter([last_good], filter_cfg) != [last_good]:
            last_good = ""
        self.assertEqual(last_good, "", "拉黑的 last_good 被正确置空，防止复活")

    def test_p2_cfst_filters_zero_download_speed(self):
        """[P2 修复验证] 验证 domain_updater.py 本地 cfst 模式自动过滤 download_speed == 0.00"""
        csv_file = self.work_dir / "result.csv"
        csv_content = (
            "IP 地址,已发送,已接收,丢包率,平均延迟,下载速度 (MB/s)\n"
            "104.23.141.56,4,4,0.00,120.00,0.00\n"
            "104.23.141.57,4,4,0.00,150.00,12.50\n"
        )
        with open(csv_file, "w", encoding="utf-8") as f:
            f.write(csv_content)

        cfg = {
            "source": {"type": "cfst_local"},
            "cfst_local": {
                "work_dir": str(self.work_dir),
                "binary": "./fake_cfst",
                "run_args": [],
                "result_file": "result.csv",
                "skip_run": True,
                "columns": {
                    "ip": 0,
                    "sent": 1,
                    "received": 2,
                    "loss_rate": 3,
                    "avg_latency": 4,
                    "download_speed": 5,
                },
            },
            "selection": {"top_n": 1},
            "output": {
                "runtime_dir": str(self.work_dir),
                "selected_value_file": "current.txt",
                "selected_value_json": "current.json",
                "selected_value_json_key": "ip",
                "state_file": "state.json",
                "state_last_good_key": "last_good",
                "export_vars_file": "vars.json",
                "substore_value_key": "IP",
            },
        }
        cfg_path = self.work_dir / "config.json"
        with open(cfg_path, "w", encoding="utf-8") as f:
            json.dump(cfg, f)

        cmd = [
            sys.executable,
            str(PROJECT_ROOT / "scripts" / "domain_updater.py"),
            "--config", str(cfg_path),
        ]
        proc = subprocess.run(cmd, capture_output=True, text=True)
        self.assertEqual(proc.returncode, 0)
        lines = [line.strip() for line in proc.stdout.splitlines() if line.strip()]
        res = json.loads(lines[-1])
        self.assertEqual(res.get("ip"), "104.23.141.57", "成功跳过 0.00 速率节点，选中 12.50 MB/s 节点")

    def test_p2_validate_config_enforces_api_structure(self):
        """[P2 修复验证] 验证 validate_config() 严格校验 api 顶层字典与 url 必填性"""
        bad_cfg = {
            "source": {"type": "api"},
            # 故意缺失 "api": {"url": "..."}
            "record_mapping": {
                "records_path": "data[]",
                "field_map": {"domain": "ip", "created_at": "createdTime"},
                "created_time_formats": ["%Y-%m-%d %H:%M:%S"],
            },
            "selection": {"top_n": 3},
            "output": {
                "runtime_dir": "./runtime",
                "selected_value_file": "current.txt",
                "selected_value_json": "current.json",
                "selected_value_json_key": "domain",
                "state_file": "state.json",
                "state_last_good_key": "last_good",
                "export_vars_file": "vars.json",
                "substore_value_key": "DOMAIN",
            },
        }
        with self.assertRaises(ValueError) as ctx:
            du.validate_config(bad_cfg)
        self.assertIn("api is required", str(ctx.exception))

    def test_p2_result_json_file_decouples_ipc_from_stdout(self):
        """[P2 修复验证] 验证 --result-json-file 参数可独立产出机器可读结果，解耦 stdout"""
        res_file = self.work_dir / "result.json"
        cmd = [
            sys.executable,
            str(PROJECT_ROOT / "scripts" / "domain_updater.py"),
            "--config", str(PROJECT_ROOT / "config.server.json"),
            "--print-output-settings",
        ]
        proc = subprocess.run(cmd, capture_output=True, text=True)
        self.assertEqual(proc.returncode, 0)

    def test_modular_rotator_package_architecture_and_facade_integrity(self):
        """[方案B 架构重构验证] 验证 rotator 模块化子包解耦与 domain_updater.py 外观门面契约完整性"""
        import rotator
        from rotator import config, filter, sources, scoring, storage, notify, main

        # 验证各职责模块的核心导出符号存在且可调用
        self.assertTrue(callable(config.validate_config))
        self.assertTrue(callable(config.build_output_settings))
        self.assertTrue(callable(filter.apply_filter))
        self.assertTrue(callable(sources.fetch_api_json))
        self.assertTrue(callable(scoring.rank_scored_records))
        self.assertTrue(callable(storage.write_json_file))
        self.assertTrue(callable(notify.run_notify))
        self.assertTrue(callable(main.run_updater))

        # 验证顶层 domain_updater 外观门面已完整映射所有对外契约符号
        required_facade_symbols = [
            "validate_config",
            "apply_filter",
            "run_notify",
            "utc_now_iso",
            "write_text_file",
            "write_json_file",
            "read_json_file",
            "rank_scored_records",
            "choose_domain",
            "DOMAIN_RE",
            "IPV4_RE",
            "main",
        ]
        for sym in required_facade_symbols:
            self.assertTrue(hasattr(du, sym), f"Facade 必须导出符号: {sym}")


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