vpn-rules/tests/test_module_spec.py

from __future__ import annotations

import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path, PurePosixPath


ROOT = Path(__file__).resolve().parents[1]
SPEC_PATH = ROOT / "MODULE_SPEC.yaml"
sys.path.insert(0, str(ROOT / "scripts"))

from gen_clash_verge_script import (  # noqa: E402
    default_output_path,
    load_payload,
    render_script,
    to_rule,
)


class ModuleSpecTests(unittest.TestCase):
    def test_clash_verge_default_output_path_is_platform_aware(self) -> None:
        home = Path("C:/Users/example")
        appdata = Path("C:/Users/example/AppData/Roaming")

        self.assertEqual(
            default_output_path("win32", home=home, appdata=appdata),
            appdata
            / "io.github.clash-verge-rev.clash-verge-rev"
            / "profiles"
            / "Script.js",
        )
        self.assertEqual(
            default_output_path("darwin", home=home),
            home
            / "Library/Application Support"
            / "io.github.clash-verge-rev.clash-verge-rev"
            / "profiles"
            / "Script.js",
        )

    def test_clash_verge_generator_supports_windows_gbk_console(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            output = Path(temp_dir) / "Script.js"
            env = os.environ.copy()
            env["PYTHONIOENCODING"] = "gbk"
            result = subprocess.run(
                [
                    sys.executable,
                    str(ROOT / "scripts/gen_clash_verge_script.py"),
                    "--out",
                    str(output),
                ],
                cwd=ROOT,
                env=env,
                capture_output=True,
                check=False,
            )

            self.assertEqual(
                result.returncode,
                0,
                result.stderr.decode("gbk", errors="replace"),
            )
            self.assertTrue(output.is_file())

    def test_module_spec_declares_rule_authority_and_safe_context(self) -> None:
        spec = json.loads(SPEC_PATH.read_text(encoding="utf-8"))

        self.assertEqual(
            set(spec),
            {
                "schema_version",
                "module_id",
                "authority",
                "repository",
                "bounded_context",
                "capabilities",
                "data",
                "contracts",
                "dependencies",
                "agent_context",
                "verification",
            },
        )
        self.assertEqual(spec["schema_version"], 1)
        self.assertEqual(spec["module_id"], "vpn-rules")
        self.assertEqual(
            spec["authority"],
            {
                "scope": "module_metadata",
                "subject": "vpn-rules",
                "owner": "vpn-rules-owner",
                "source": "MODULE_SPEC.yaml",
                "revision": 1,
                "effective_from": "2026-08-20T00:00:00+08:00",
            },
        )
        self.assertEqual(
            spec["repository"],
            {
                "name": "vpn-rules",
                "workspace_id": None,
                "type": "infrastructure",
                "maturity": "operational",
            },
        )

        boundary = spec["bounded_context"]
        self.assertEqual(boundary["domain"], "network-routing-policy")
        self.assertIn("single source of truth", boundary["responsibility"].lower())
        prohibited = " ".join(boundary["prohibited_responsibilities"]).lower()
        self.assertIn("production", prohibited)
        self.assertIn("credential", prohibited)
        self.assertIn("proxy subscription", prohibited)

        self.assertEqual(
            {capability["id"] for capability in spec["capabilities"]},
            {
                "direct-routing-rule-authoring",
                "clash-verge-script-generation",
                "router-rule-projection",
            },
        )
        self.assertEqual(spec["contracts"]["consumes"], [])
        self.assertEqual(spec["dependencies"], [])
        provided = {
            contract["contract_id"]: contract
            for contract in spec["contracts"]["provides"]
        }
        self.assertEqual(
            set(provided),
            {
                "mihomo-direct-rule-provider",
                "router-direct-rules",
                "router-direct-plain-list",
            },
        )
        self.assertEqual(
            {contract["version"] for contract in provided.values()}, {"1.0.0"}
        )
        for contract in provided.values():
            self.assertTrue((ROOT / contract["path"]).is_file())

        context = spec["agent_context"]
        self.assertLessEqual(context["max_default_tokens"], 6000)
        self.assertEqual(
            {entry["path"] for entry in context["default_entrypoints"]},
            {"README.md", "direct.yaml"},
        )
        for entry in context["default_entrypoints"]:
            self.assertTrue((ROOT / entry["path"]).is_file())
        self.assertIn("router", context["excluded_paths"])
        for value in context["excluded_paths"]:
            path = PurePosixPath(value)
            self.assertFalse(path.is_absolute())
            self.assertNotIn("..", path.parts)

        self.assertEqual(
            spec["verification"],
            {
                "commands": [
                    {
                        "id": "rule-contract-tests",
                        "argv": [
                            "python3",
                            "-m",
                            "unittest",
                            "discover",
                            "-s",
                            "tests",
                            "-v",
                        ],
                        "cwd": ".",
                        "network": False,
                        "required": True,
                    }
                ]
            },
        )

    def test_checked_in_router_projections_match_the_rule_source(self) -> None:
        payload = load_payload(ROOT / "direct.yaml")
        plain = (ROOT / "router/direct-plain.txt").read_text(encoding="utf-8")
        router = (ROOT / "router/direct-rules.yaml").read_text(encoding="utf-8")

        self.assertEqual(plain, "\n".join(payload) + "\n")
        self.assertEqual(
            router.split("rules:\n", 1)[1],
            "\n".join(f"  - {to_rule(entry)}" for entry in payload) + "\n",
        )
        self.assertTrue(all(",DIRECT" not in entry for entry in payload))

    def test_onedrive_process_is_forced_to_use_direct_routing(self) -> None:
        payload = load_payload(ROOT / "direct.yaml")

        self.assertIn("PROCESS-NAME,OneDrive.exe", payload)
        self.assertEqual(
            to_rule("PROCESS-NAME,OneDrive.exe"),
            "PROCESS-NAME,OneDrive.exe,DIRECT",
        )

    def test_clash_verge_script_adds_reachable_direct_dns_fallbacks(self) -> None:
        script = render_script(["PROCESS-NAME,OneDrive.exe"])

        self.assertIn("const directDnsFallbacks", script)
        self.assertIn("https://dns.alidns.com/dns-query", script)
        self.assertIn("https://doh.pub/dns-query", script)
        self.assertIn("config.dns = dns", script)


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