automtu/tests/unit/test_core.py

import io
import pathlib
import tempfile
import unittest
from contextlib import redirect_stdout
from types import SimpleNamespace
from unittest.mock import patch

from automtu import net
from automtu.core import run_automtu


def _args(**over) -> SimpleNamespace:
    base = {
        "dry_run": True,
        "egress_if": None,
        "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,
        "apply_docker_mtu": False,
        "apply_all": False,
        "docker_if": None,
        "docker_no_user_bridges": False,
        "wg_if": "wg0",
        "wg_overhead": 80,
        "wg_min": 1280,
        "set_wg_mtu": None,
        "persist": None,
        "uninstall": False,
        "print_mtu": None,
        "print_json": False,
    }
    base.update(over)
    return SimpleNamespace(**base)


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,
            apply_docker_mtu=False,
            apply_all=False,
            docker_if=None,
            docker_no_user_bridges=False,
            wg_if="wg0",
            wg_overhead=80,
            wg_min=1280,
            set_wg_mtu=None,
            persist=None,
            uninstall=False,
            print_mtu=None,
            print_json=False,
        )

        # 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),
            patch("automtu.core.detect_docker_ifaces", return_value=[]),
        ):
            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)

        mock_set.assert_any_call("eth0", 1452, True)
        mock_set.assert_any_call("wg0", 1372, True)

    def test_run_automtu_apply_all_includes_docker_bridge(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,
            apply_docker_mtu=False,
            apply_all=True,  # expands in core()
            docker_if=None,
            docker_no_user_bridges=False,
            wg_if="wg0",
            wg_overhead=80,
            wg_min=1280,
            set_wg_mtu=None,
            persist=None,
            uninstall=False,
            print_mtu=None,
            print_json=False,
        )

        with (
            patch("automtu.core.require_root", return_value=None),
            patch(
                "automtu.core.iface_exists",
                side_effect=lambda name: name in {"eth0", "wg0", "docker0", "br-abc"},
            ),
            patch("automtu.core.read_iface_mtu", return_value=1500),
            patch("automtu.core.set_iface_mtu") as mock_set,
            patch("automtu.core.wg_is_active", return_value=True),
            patch(
                "automtu.core.detect_docker_ifaces", return_value=["docker0", "br-abc"]
            ),
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = run_automtu(args)

        self.assertEqual(rc, 0)

        # egress applied
        mock_set.assert_any_call("eth0", 1500, True)

        # wg applied (1500-80=1420)
        mock_set.assert_any_call("wg0", 1420, True)

        # docker applied
        mock_set.assert_any_call("docker0", 1500, True)
        mock_set.assert_any_call("br-abc", 1500, 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,
            apply_docker_mtu=False,
            apply_all=False,
            docker_if=None,
            docker_no_user_bridges=False,
            wg_if="wg0",
            wg_overhead=80,
            wg_min=1280,
            set_wg_mtu=None,
            persist=None,
            uninstall=False,
            print_mtu=None,
            print_json=False,
        )

        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,
            patch("automtu.core.detect_docker_ifaces", return_value=[]),
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = run_automtu(args)

        self.assertEqual(rc, 0)
        mock_set.assert_not_called()


class TestCoreInSysboxContainer(unittest.TestCase):
    """Container with dead /sys/class/net symlinks: routing and ip link work."""

    def setUp(self) -> None:
        tmp = tempfile.TemporaryDirectory()
        self.addCleanup(tmp.cleanup)
        root = pathlib.Path(tmp.name)
        self.netdir = root / "class" / "net"
        self.netdir.mkdir(parents=True)
        for name in ("eth0", "lo"):
            (self.netdir / name).symlink_to(root / "devices" / "virtual" / "net" / name)

    @staticmethod
    def _fake_run(cmd: list[str]) -> str:
        if cmd[:5] == ["ip", "-4", "route", "show", "default"]:
            return "default via 172.28.0.1 dev eth0"
        if cmd[:3] == ["ip", "link", "show"] and cmd[-1] == "eth0":
            return (
                "2: eth0@if5: <BROADCAST,MULTICAST,UP> mtu 1450 qdisc noqueue state UP"
            )
        return ""

    def test_print_mtu_effective_yields_number_and_rc0(self) -> None:
        with (
            patch.object(net, "SYSFS_NET", self.netdir),
            patch.object(net, "_run", side_effect=self._fake_run),
            patch("automtu.core.probe_pmtu", return_value=1400),
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = run_automtu(_args(pmtu_target=["1.1.1.1"], print_mtu="effective"))

        self.assertEqual(rc, 0)
        self.assertEqual(int(buf.getvalue().strip()), 1400)

    def test_explicit_egress_if_is_accepted(self) -> None:
        with (
            patch.object(net, "SYSFS_NET", self.netdir),
            patch.object(net, "_run", side_effect=self._fake_run),
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                rc = run_automtu(_args(egress_if="eth0", print_mtu="egress"))

        self.assertEqual(rc, 0)
        self.assertEqual(int(buf.getvalue().strip()), 1450)


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