77 lines
2.7 KiB
Python
77 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from vpn_egressctl.errors import ValidationError
|
|
from vpn_egressctl.policy import load_policy
|
|
|
|
|
|
class PolicyTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.raw = json.loads(Path("config/policy.json").read_text(encoding="utf-8"))
|
|
|
|
def load(self, raw: dict) -> object:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "policy.json"
|
|
path.write_text(json.dumps(raw), encoding="utf-8")
|
|
return load_policy(path)
|
|
|
|
def test_production_policy(self) -> None:
|
|
policy = self.load(self.raw)
|
|
self.assertEqual(policy.sing_box.required_version, "1.14.0")
|
|
self.assertEqual(policy.network.iproute2_table_index, 2022)
|
|
|
|
def test_unknown_root_key(self) -> None:
|
|
self.raw["future"] = True
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_unknown_nested_key(self) -> None:
|
|
self.raw["network"]["typo"] = 1
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_missing_key(self) -> None:
|
|
del self.raw["dns"]["strategy"]
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_version_is_exactly_pinned(self) -> None:
|
|
for value in ("1.13.19", "1.14", ">=1.14,<1.15", "1.14.1", "1.15.0"):
|
|
raw = json.loads(json.dumps(self.raw))
|
|
raw["sing_box"]["required_version"] = value
|
|
with self.subTest(value=value), self.assertRaises(ValidationError):
|
|
self.load(raw)
|
|
|
|
def test_public_exclusion_is_valid_but_diagnosable(self) -> None:
|
|
self.raw["network"]["route_exclude_address"].append("203.0.113.1/32")
|
|
policy = self.load(self.raw)
|
|
self.assertIn("203.0.113.1/32", policy.network.route_exclude_address)
|
|
|
|
def test_invalid_network_rejected(self) -> None:
|
|
self.raw["network"]["route_exclude_address"] = ["10.20.0.1/24"]
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_insecure_health_url_rejected(self) -> None:
|
|
self.raw["healthcheck"]["url"] = "http://example.com/"
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_health_url_credentials_rejected(self) -> None:
|
|
self.raw["healthcheck"]["url"] = "https://user:pass@example.com/check"
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
def test_dns_path_must_be_absolute(self) -> None:
|
|
self.raw["dns"]["remote_path"] = "dns-query"
|
|
with self.assertRaises(ValidationError):
|
|
self.load(self.raw)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|