refactor: convert script to automtu package with CI workflow
https://chatgpt.com/share/697112b2-0410-800f-93ff-9372b603d43f
This commit is contained in:
0
tests/unit/__init__.py
Normal file
0
tests/unit/__init__.py
Normal file
14
tests/unit/test___init__.py
Normal file
14
tests/unit/test___init__.py
Normal file
@@ -0,0 +1,14 @@
|
||||
import unittest
|
||||
|
||||
import automtu
|
||||
|
||||
|
||||
class TestInit(unittest.TestCase):
|
||||
def test_version_is_exposed(self) -> None:
|
||||
self.assertTrue(hasattr(automtu, "__version__"))
|
||||
self.assertIsInstance(automtu.__version__, str)
|
||||
self.assertRegex(automtu.__version__, r"^\d+\.\d+\.\d+$")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
26
tests/unit/test___main__.py
Normal file
26
tests/unit/test___main__.py
Normal file
@@ -0,0 +1,26 @@
|
||||
import unittest
|
||||
from unittest.mock import Mock, patch
|
||||
|
||||
import automtu.__main__ as entry
|
||||
|
||||
|
||||
class TestMain(unittest.TestCase):
|
||||
def test_main_calls_parser_and_core(self) -> None:
|
||||
fake_args = object()
|
||||
fake_parser = Mock()
|
||||
fake_parser.parse_args.return_value = fake_args
|
||||
|
||||
with (
|
||||
patch("automtu.__main__.build_parser", return_value=fake_parser) as p_build,
|
||||
patch("automtu.__main__.run_automtu", return_value=0) as p_run,
|
||||
):
|
||||
rc = entry.main()
|
||||
|
||||
self.assertEqual(rc, 0)
|
||||
p_build.assert_called_once_with()
|
||||
fake_parser.parse_args.assert_called_once_with()
|
||||
p_run.assert_called_once_with(fake_args)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
62
tests/unit/test_cli.py
Normal file
62
tests/unit/test_cli.py
Normal file
@@ -0,0 +1,62 @@
|
||||
import unittest
|
||||
|
||||
from automtu.cli import build_parser
|
||||
|
||||
|
||||
class TestCli(unittest.TestCase):
|
||||
def test_build_parser_accepts_known_args(self) -> None:
|
||||
p = build_parser()
|
||||
|
||||
# Parse only; no side effects.
|
||||
args = p.parse_args(
|
||||
[
|
||||
"--egress-if",
|
||||
"eth0",
|
||||
"--prefer-wg-egress",
|
||||
"--pmtu-target",
|
||||
"1.1.1.1,8.8.8.8",
|
||||
"--pmtu-timeout",
|
||||
"2.0",
|
||||
"--pmtu-min-payload",
|
||||
"1200",
|
||||
"--pmtu-max-payload",
|
||||
"1472",
|
||||
"--pmtu-policy",
|
||||
"median",
|
||||
"--apply-egress-mtu",
|
||||
"--apply-wg-mtu",
|
||||
"--wg-if",
|
||||
"wg0",
|
||||
"--wg-overhead",
|
||||
"80",
|
||||
"--wg-min",
|
||||
"1280",
|
||||
"--auto-pmtu-from-wg",
|
||||
"--set-wg-mtu",
|
||||
"1372",
|
||||
"--force-egress-mtu",
|
||||
"1452",
|
||||
"--dry-run",
|
||||
]
|
||||
)
|
||||
|
||||
self.assertEqual(args.egress_if, "eth0")
|
||||
self.assertTrue(args.prefer_wg_egress)
|
||||
self.assertEqual(args.pmtu_target, ["1.1.1.1,8.8.8.8"])
|
||||
self.assertEqual(args.pmtu_timeout, 2.0)
|
||||
self.assertEqual(args.pmtu_min_payload, 1200)
|
||||
self.assertEqual(args.pmtu_max_payload, 1472)
|
||||
self.assertEqual(args.pmtu_policy, "median")
|
||||
self.assertTrue(args.apply_egress_mtu)
|
||||
self.assertTrue(args.apply_wg_mtu)
|
||||
self.assertEqual(args.wg_if, "wg0")
|
||||
self.assertEqual(args.wg_overhead, 80)
|
||||
self.assertEqual(args.wg_min, 1280)
|
||||
self.assertTrue(args.auto_pmtu_from_wg)
|
||||
self.assertEqual(args.set_wg_mtu, 1372)
|
||||
self.assertEqual(args.force_egress_mtu, 1452)
|
||||
self.assertTrue(args.dry_run)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
94
tests/unit/test_core.py
Normal file
94
tests/unit/test_core.py
Normal file
@@ -0,0 +1,94 @@
|
||||
import io
|
||||
import unittest
|
||||
from contextlib import redirect_stdout
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import patch
|
||||
|
||||
from automtu.core import run_automtu
|
||||
|
||||
|
||||
class TestCore(unittest.TestCase):
|
||||
def test_run_automtu_happy_path_all_mocked(self) -> None:
|
||||
args = SimpleNamespace(
|
||||
dry_run=True,
|
||||
egress_if=None,
|
||||
prefer_wg_egress=False,
|
||||
force_egress_mtu=None,
|
||||
pmtu_target=["1.1.1.1,8.8.8.8"],
|
||||
auto_pmtu_from_wg=False,
|
||||
pmtu_min_payload=1200,
|
||||
pmtu_max_payload=1472,
|
||||
pmtu_timeout=1.0,
|
||||
pmtu_policy="min",
|
||||
apply_egress_mtu=True,
|
||||
apply_wg_mtu=True,
|
||||
wg_if="wg0",
|
||||
wg_overhead=80,
|
||||
wg_min=1280,
|
||||
set_wg_mtu=None,
|
||||
)
|
||||
|
||||
# PMTU probes: 1452 and 1500 -> min policy => 1452, effective=min(base(1500),1452)=1452
|
||||
# wg_mtu = 1452-80 = 1372
|
||||
with (
|
||||
patch("automtu.core.require_root", return_value=None),
|
||||
patch("automtu.core.detect_egress_iface", return_value="eth0"),
|
||||
patch("automtu.core.iface_exists", return_value=True),
|
||||
patch("automtu.core.read_iface_mtu", return_value=1500),
|
||||
patch("automtu.core.probe_pmtu", side_effect=[1452, 1500]),
|
||||
patch("automtu.core.set_iface_mtu") as mock_set,
|
||||
patch("automtu.core.wg_is_active", return_value=False),
|
||||
patch("automtu.core.wg_peer_endpoints", return_value=[]),
|
||||
patch("automtu.core.default_route_uses_iface", return_value=False),
|
||||
):
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
rc = run_automtu(args)
|
||||
|
||||
self.assertEqual(rc, 0)
|
||||
s = buf.getvalue()
|
||||
self.assertIn("Detected egress interface: eth0", s)
|
||||
self.assertIn("Egress base MTU: 1500", s)
|
||||
self.assertIn("Selected Path MTU (policy=min): 1452", s)
|
||||
self.assertIn("Computed wg0 MTU: 1372", s)
|
||||
|
||||
# apply egress and wg are both true -> two calls in order
|
||||
mock_set.assert_any_call("eth0", 1452, True)
|
||||
mock_set.assert_any_call("wg0", 1372, True)
|
||||
|
||||
def test_run_automtu_does_not_apply_wg_without_flag(self) -> None:
|
||||
args = SimpleNamespace(
|
||||
dry_run=True,
|
||||
egress_if="eth0",
|
||||
prefer_wg_egress=False,
|
||||
force_egress_mtu=None,
|
||||
pmtu_target=None,
|
||||
auto_pmtu_from_wg=False,
|
||||
pmtu_min_payload=1200,
|
||||
pmtu_max_payload=1472,
|
||||
pmtu_timeout=1.0,
|
||||
pmtu_policy="min",
|
||||
apply_egress_mtu=False,
|
||||
apply_wg_mtu=False,
|
||||
wg_if="wg0",
|
||||
wg_overhead=80,
|
||||
wg_min=1280,
|
||||
set_wg_mtu=None,
|
||||
)
|
||||
|
||||
with (
|
||||
patch("automtu.core.require_root", return_value=None),
|
||||
patch("automtu.core.iface_exists", return_value=True),
|
||||
patch("automtu.core.read_iface_mtu", return_value=1500),
|
||||
patch("automtu.core.set_iface_mtu") as mock_set,
|
||||
):
|
||||
buf = io.StringIO()
|
||||
with redirect_stdout(buf):
|
||||
rc = run_automtu(args)
|
||||
|
||||
self.assertEqual(rc, 0)
|
||||
mock_set.assert_not_called()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
49
tests/unit/test_net.py
Normal file
49
tests/unit/test_net.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import automtu.net as net
|
||||
|
||||
|
||||
class TestNet(unittest.TestCase):
|
||||
def test_detect_egress_iface_parses_default_route_and_ignores_vpn(self) -> None:
|
||||
# Simulate: ip route show default -> dev wg0 first, then eth0
|
||||
ip4_default = (
|
||||
"default via 10.0.0.1 dev wg0 proto dhcp src 10.0.0.2 metric 100\n"
|
||||
)
|
||||
ip6_default = "default via fe80::1 dev eth0 proto ra metric 100\n"
|
||||
|
||||
def fake_run(cmd: list[str]) -> str:
|
||||
if cmd[:4] == ["ip", "-4", "route", "show"]:
|
||||
return ip4_default.strip()
|
||||
if cmd[:4] == ["ip", "-6", "route", "show"]:
|
||||
return ip6_default.strip()
|
||||
return ""
|
||||
|
||||
with (
|
||||
patch("automtu.net._run", side_effect=fake_run),
|
||||
patch("automtu.net.iface_exists", return_value=True),
|
||||
):
|
||||
# ignore_vpn=True -> should skip wg0 and pick eth0
|
||||
self.assertEqual(net.detect_egress_iface(ignore_vpn=True), "eth0")
|
||||
|
||||
# ignore_vpn=False -> first seen is wg0
|
||||
self.assertEqual(net.detect_egress_iface(ignore_vpn=False), "wg0")
|
||||
|
||||
def test_default_route_uses_iface_true_false(self) -> None:
|
||||
ip4_default = "default via 10.0.0.1 dev eth0\n"
|
||||
ip6_default = ""
|
||||
|
||||
def fake_run(cmd: list[str]) -> str:
|
||||
if cmd[:4] == ["ip", "-4", "route", "show"]:
|
||||
return ip4_default.strip()
|
||||
if cmd[:4] == ["ip", "-6", "route", "show"]:
|
||||
return ip6_default.strip()
|
||||
return ""
|
||||
|
||||
with patch("automtu.net._run", side_effect=fake_run):
|
||||
self.assertTrue(net.default_route_uses_iface("eth0"))
|
||||
self.assertFalse(net.default_route_uses_iface("wg0"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
36
tests/unit/test_pmtu.py
Normal file
36
tests/unit/test_pmtu.py
Normal file
@@ -0,0 +1,36 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import automtu.pmtu as pmtu
|
||||
|
||||
|
||||
class TestPmtu(unittest.TestCase):
|
||||
def test_probe_pmtu_binary_search_and_hdr_addition(self) -> None:
|
||||
# Mock _is_ipv6 -> IPv4, so hdr = 28.
|
||||
# Mock _ping_ok so that payload <= 1400 works, >1400 fails.
|
||||
def fake_ping_ok(payload: int, target: str, timeout_s: float) -> bool:
|
||||
return payload <= 1400
|
||||
|
||||
with (
|
||||
patch("automtu.pmtu._is_ipv6", return_value=False),
|
||||
patch("automtu.pmtu._ping_ok", side_effect=fake_ping_ok),
|
||||
):
|
||||
# lo=1200 works, hi=1472 partially works -> best = 1400 -> mtu = 1400+28 = 1428
|
||||
mtu = pmtu.probe_pmtu(
|
||||
"1.1.1.1", lo_payload=1200, hi_payload=1472, timeout=1.0
|
||||
)
|
||||
self.assertEqual(mtu, 1428)
|
||||
|
||||
def test_probe_pmtu_returns_none_if_even_floor_fails(self) -> None:
|
||||
with (
|
||||
patch("automtu.pmtu._is_ipv6", return_value=False),
|
||||
patch("automtu.pmtu._ping_ok", return_value=False),
|
||||
):
|
||||
mtu = pmtu.probe_pmtu(
|
||||
"1.1.1.1", lo_payload=1200, hi_payload=1472, timeout=1.0
|
||||
)
|
||||
self.assertIsNone(mtu)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
58
tests/unit/test_wg.py
Normal file
58
tests/unit/test_wg.py
Normal file
@@ -0,0 +1,58 @@
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import automtu.wg as wg
|
||||
|
||||
|
||||
class TestWg(unittest.TestCase):
|
||||
def test_wg_peer_endpoints_from_wg_show(self) -> None:
|
||||
# wg show wg0 endpoints output
|
||||
out = "abcde12345\t46.4.224.77:51820\nfffff00000\t[2a01:db8::1]:51820\n"
|
||||
|
||||
with (
|
||||
patch(
|
||||
"automtu.wg._run",
|
||||
side_effect=lambda cmd: out if cmd[:3] == ["wg", "show", "wg0"] else "",
|
||||
),
|
||||
patch("automtu.wg.iface_exists", return_value=True),
|
||||
patch("automtu.wg._rc", return_value=0),
|
||||
):
|
||||
eps = wg.wg_peer_endpoints("wg0")
|
||||
|
||||
self.assertEqual(eps, ["46.4.224.77", "2a01:db8::1"])
|
||||
|
||||
def test_wg_peer_endpoints_fallback_showconf(self) -> None:
|
||||
show_endpoints_empty = ""
|
||||
showconf = (
|
||||
"[Interface]\n"
|
||||
"PrivateKey = x\n"
|
||||
"[Peer]\n"
|
||||
"Endpoint = 46.4.224.77:51820\n"
|
||||
"[Peer]\n"
|
||||
"Endpoint = [2a01:db8::1]:51820\n"
|
||||
)
|
||||
|
||||
def fake_run(cmd: list[str]) -> str:
|
||||
if cmd == ["wg", "show", "wg0", "endpoints"]:
|
||||
return show_endpoints_empty
|
||||
if cmd == ["wg", "showconf", "wg0"]:
|
||||
return showconf
|
||||
return ""
|
||||
|
||||
with patch("automtu.wg._run", side_effect=fake_run):
|
||||
eps = wg.wg_peer_endpoints("wg0")
|
||||
|
||||
self.assertEqual(eps, ["46.4.224.77", "2a01:db8::1"])
|
||||
|
||||
def test_wg_is_active_uses_iface_exists_and_wg_show_rc(self) -> None:
|
||||
with (
|
||||
patch("automtu.wg.iface_exists", return_value=True),
|
||||
patch("automtu.wg._rc", return_value=0),
|
||||
):
|
||||
self.assertTrue(wg.wg_is_active("wg0"))
|
||||
with patch("automtu.wg.iface_exists", return_value=False):
|
||||
self.assertFalse(wg.wg_is_active("wg0"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user