62 lines
2.6 KiB
Python
62 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from vpn_egressctl.renderer_1_13_19 import render_bytes, render_config
|
|
from vpn_egressctl.uri import parse_hysteria2_uri
|
|
|
|
from tests.helpers import make_policy
|
|
|
|
|
|
class RendererTests(unittest.TestCase):
|
|
def test_complete_production_shape(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
policy = make_policy(Path(directory))
|
|
endpoint = parse_hysteria2_uri(
|
|
"hysteria2://AUTH@fi.api.withen.pro:443/?insecure=0&obfs=salamander&obfs-password=OBFS"
|
|
)
|
|
config = render_config(policy, endpoint)
|
|
self.assertEqual(list(config), ["log", "dns", "inbounds", "outbounds", "route"])
|
|
tun = config["inbounds"][0]
|
|
self.assertEqual(tun["route_exclude_address"], ["10.20.0.0/24", "10.30.0.0/24", "127.0.0.0/8"])
|
|
self.assertEqual(tun["iproute2_table_index"], 2022)
|
|
self.assertEqual(tun["auto_redirect_input_mark"], "0x2023")
|
|
self.assertEqual(tun["auto_redirect_output_mark"], "0x2024")
|
|
self.assertEqual(tun["auto_redirect_reset_mark"], "0x2025")
|
|
outbound = config["outbounds"][0]
|
|
self.assertEqual(outbound["server"], "fi.api.withen.pro")
|
|
self.assertEqual(outbound["password"], "AUTH")
|
|
self.assertEqual(outbound["obfs"]["password"], "OBFS")
|
|
self.assertEqual(outbound["bind_interface"], "eth0")
|
|
self.assertEqual([item["tag"] for item in config["outbounds"]], ["hy2-out"])
|
|
self.assertEqual(config["dns"]["servers"][1]["detour"], "hy2-out")
|
|
self.assertEqual(config["route"]["final"], "hy2-out")
|
|
self.assertNotIn("185.156.108.141", render_bytes(policy, endpoint).decode())
|
|
|
|
def test_multi_port_mapping(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
config = render_config(
|
|
make_policy(Path(directory)),
|
|
parse_hysteria2_uri("hy2://x@example.com:443,5000-6000"),
|
|
)
|
|
outbound = config["outbounds"][0]
|
|
self.assertNotIn("server_port", outbound)
|
|
self.assertEqual(outbound["server_ports"], ["443", "5000:6000"])
|
|
|
|
def test_deterministic_utf8_json(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
policy = make_policy(Path(directory))
|
|
endpoint = parse_hysteria2_uri("hy2://x@example.com#Тест")
|
|
first = render_bytes(policy, endpoint)
|
|
second = render_bytes(policy, endpoint)
|
|
self.assertEqual(first, second)
|
|
self.assertTrue(first.endswith(b"\n"))
|
|
json.loads(first)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|