automtu/tests/unit/test_net.py

import pathlib
import tempfile
import unittest
from unittest.mock import patch

from automtu import net

IP_LINK_ETH0 = (
    "2: eth0@if5: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1420 qdisc noqueue state UP "
    "mode DEFAULT group default\n    link/ether 02:42:ac:1c:00:02 brd ff:ff:ff:ff:ff:ff"
)
IP_LINK_ALL = (
    "1: lo: <LOOPBACK,UP,LOWER_UP> mtu 65536 qdisc noqueue state UNKNOWN\n"
    "2: eth0@if5: <BROADCAST,MULTICAST,UP,LOWER_UP> mtu 1420 qdisc noqueue state UP"
)


def _sysbox_sysfs(root: pathlib.Path, ifaces=("eth0", "lo")) -> pathlib.Path:
    """/sys replacement as seen inside a sysbox container: dead symlinks."""
    netdir = root / "class" / "net"
    netdir.mkdir(parents=True)
    for name in ifaces:
        (netdir / name).symlink_to(root / "devices" / "virtual" / "net" / name)
    return netdir


class SysboxSysfsBase(unittest.TestCase):
    def setUp(self) -> None:
        self._tmp = tempfile.TemporaryDirectory()
        self.root = pathlib.Path(self._tmp.name)
        self.addCleanup(self._tmp.cleanup)


class TestDeadSymlinkSysfs(SysboxSysfsBase):
    def test_iface_exists_true_for_dead_symlink_without_netlink(self) -> None:
        netdir = _sysbox_sysfs(self.root)
        self.assertFalse((netdir / "eth0").exists())

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", return_value=""),
        ):
            self.assertTrue(net.iface_exists("eth0"))
            self.assertFalse(net.iface_exists("eth9"))

    def test_iface_exists_falls_back_to_netlink_without_sysfs(self) -> None:
        with (
            patch.object(net, "SYSFS_NET", self.root / "absent" / "net"),
            patch.object(net, "_run", return_value=IP_LINK_ETH0),
        ):
            self.assertTrue(net.iface_exists("eth0"))

    def test_detect_egress_iface_accepts_dead_symlink_iface(self) -> None:
        netdir = _sysbox_sysfs(self.root)

        def fake_run(cmd: list[str]) -> str:
            if cmd[:5] == ["ip", "-4", "route", "show", "default"]:
                return "default via 172.28.0.1 dev eth0"
            return ""

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", side_effect=fake_run),
        ):
            self.assertEqual(net.detect_egress_iface(), "eth0")

    def test_list_ifaces_keeps_dead_symlinks_and_skips_plain_files(self) -> None:
        netdir = _sysbox_sysfs(self.root, ("eth0", "lo", "br-abc123"))
        (netdir / "bonding_masters").write_text("")

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", return_value=""),
        ):
            self.assertEqual(net.list_ifaces(), ["br-abc123", "eth0", "lo"])

    def test_list_ifaces_falls_back_to_netlink_without_sysfs(self) -> None:
        with (
            patch.object(net, "SYSFS_NET", self.root / "absent" / "net"),
            patch.object(net, "_run", return_value=IP_LINK_ALL),
        ):
            self.assertEqual(net.list_ifaces(), ["eth0", "lo"])


class TestReadIfaceMtu(SysboxSysfsBase):
    def test_prefers_sysfs_over_netlink(self) -> None:
        netdir = self.root / "class" / "net"
        (netdir / "eth0").mkdir(parents=True)
        (netdir / "eth0" / "mtu").write_text("1500\n")

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", return_value=IP_LINK_ETH0) as run,
        ):
            self.assertEqual(net.read_iface_mtu("eth0"), 1500)
            run.assert_not_called()

    def test_netlink_fallback_on_dead_symlink(self) -> None:
        netdir = _sysbox_sysfs(self.root)

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", return_value=IP_LINK_ETH0),
        ):
            self.assertEqual(net.read_iface_mtu("eth0"), 1420)

    def test_raises_when_neither_sysfs_nor_netlink_answers(self) -> None:
        netdir = _sysbox_sysfs(self.root)

        with (
            patch.object(net, "SYSFS_NET", netdir),
            patch.object(net, "_run", return_value=""),
            self.assertRaises(RuntimeError) as ctx,
        ):
            net.read_iface_mtu("eth0")

        self.assertIn("eth0", str(ctx.exception))


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"))

    def test_list_ifaces_returns_sorted_names(self) -> None:
        with tempfile.TemporaryDirectory() as tmp:
            netdir = _sysbox_sysfs(pathlib.Path(tmp), ("wg0", "eth0", "lo"))
            with patch.object(net, "SYSFS_NET", netdir):
                self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])


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