4 Commits

Author SHA1 Message Date
8378591520 Release version 1.1.1
Some checks failed
CI / test-and-lint (push) Has been cancelled
2026-09-08 19:39:40 +02:00
75c1164ad7 chore(claude): ask before touching CHANGELOG.md and pyproject.toml
Version bumps and changelog entries are release decisions, so an agent
should not write those two files on its own. Project-scoped ask rules make
Claude Code prompt instead.

Scope note: these rules match the Edit and Write tools, which are
path-based. Bash is matched by command prefix, not by path, so a shell
write (sed -i, a heredoc) is not covered.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 18:44:37 +02:00
6392131577 style: satisfy the widened ruff default rule set
ruff 0.16 enables rules that earlier releases did not, so `ruff check .`
was already failing on HEAD with 35 findings - the CI job installs ruff
unpinned and would have gone red on the next push regardless of this
branch.

All changes are mechanical and behaviour-neutral:

- Optional[X] -> X | None, List[str] -> list[str], dropped the now unused
  typing imports (every module already carries
  `from __future__ import annotations`, and requires-python is >= 3.10).
- Sorted import blocks, `import automtu.x as x` -> `from automtu import x`.
- subprocess.run in pmtu.py and wg.py now passes check=False explicitly.
  That is the parameter default, so the calls behave exactly as before.
- int(round(x)) -> round(x): round() with a single argument already
  returns int.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 18:42:55 +02:00
499e61346d fix(net): detect interfaces without following /sys symlinks
/sys/class/net/<if> is always a symlink into /sys/devices; in nested
containers (sysbox runtime) that target is not visible, so the link is
dead. iface_exists() used Path.exists(), which follows the link and
therefore tested the visibility of the target instead of the existence of
the interface. detect_egress_iface() discarded every interface it had
correctly read from the routing table, so automtu aborted with "Could not
detect egress interface" and rc=2 even though `ip -4 route show default`
and `ip link show dev eth0` both worked. The documented escape hatch
--egress-if was equally dead, because core.py validates it through the
same call.

Chosen fix: os.path.lexists() rather than probing netlink for existence.
It asks the right question -- "is there an entry named <if>" -- and costs
no subprocess on a healthy host. Netlink (`ip link show`) is only the
fallback for when /sys/class/net itself is unavailable, so sysfs stays the
preferred path everywhere.

read_iface_mtu() gains the same cascade: sysfs first, MTU parsed from
`ip link show dev <if>` when the sysfs path is unreadable, RuntimeError
naming both sources when neither answers - no silent default. core.py
turns that into an error line plus rc=3 instead of a traceback.

list_ifaces() had the same defect (is_dir() on a dead symlink) and would
have left Docker bridge detection blind in the same environments.

Regression tests build a /sys replacement whose class/net/eth0 points at a
missing target and mock the ip command: they fail if lexists becomes
exists again, if the MTU fallback is removed, or if the symlink filter is
dropped.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-09-08 18:35:25 +02:00
17 changed files with 296 additions and 67 deletions

10
.claude/settings.json Normal file
View File

@@ -0,0 +1,10 @@
{
"permissions": {
"ask": [
"Edit(CHANGELOG.md)",
"Write(CHANGELOG.md)",
"Edit(pyproject.toml)",
"Write(pyproject.toml)"
]
}
}

View File

@@ -1,3 +1,12 @@
# Changelog
## [1.1.1] - 2026-09-08
* Fixed interface detection with dead /sys/class/net symlinks (sysbox)
* Added netlink fallback for reading MTU when sysfs is unreadable
* Fixed Docker bridge detection in the same environments
* Fixed lint against the widened ruff default rule set
## [1.1.0] - 2026-01-23 ## [1.1.0] - 2026-01-23
* * Added persistent MTU configuration via systemd (install & uninstall) * * Added persistent MTU configuration via systemd (install & uninstall)

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "automtu" name = "automtu"
version = "1.1.0" version = "1.1.1"
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)." description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
readme = "README.md" readme = "README.md"
requires-python = ">=3.10" requires-python = ">=3.10"

View File

@@ -2,8 +2,8 @@ from __future__ import annotations
import statistics import statistics
import sys import sys
from collections.abc import Iterable
from dataclasses import dataclass from dataclasses import dataclass
from typing import Iterable, Optional
from .docker import detect_docker_ifaces from .docker import detect_docker_ifaces
from .net import ( from .net import (
@@ -28,7 +28,7 @@ class Result:
wg_mtu: int wg_mtu: int
def _split_targets(items: Optional[list[str]]) -> list[str]: def _split_targets(items: list[str] | None) -> list[str]:
raw: list[str] = [] raw: list[str] = []
for item in items or []: for item in items or []:
raw.extend([x.strip() for x in item.split(",") if x.strip()]) raw.extend([x.strip() for x in item.split(",") if x.strip()])
@@ -119,7 +119,11 @@ def run_automtu(args) -> int:
set_iface_mtu(egress, args.force_egress_mtu, args.dry_run) set_iface_mtu(egress, args.force_egress_mtu, args.dry_run)
base_mtu = int(args.force_egress_mtu) base_mtu = int(args.force_egress_mtu)
else: else:
base_mtu = int(read_iface_mtu(egress)) try:
base_mtu = int(read_iface_mtu(egress))
except RuntimeError as exc:
print(f"[automtu][ERROR] {exc}", file=sys.stderr)
return 3
log(f"[automtu] Egress base MTU: {base_mtu}") log(f"[automtu] Egress base MTU: {base_mtu}")
# Targets (explicit + optional WG auto targets) # Targets (explicit + optional WG auto targets)
@@ -140,8 +144,8 @@ def run_automtu(args) -> int:
# PMTU probing # PMTU probing
effective_mtu = base_mtu effective_mtu = base_mtu
probe_results: dict[str, Optional[int]] = {} probe_results: dict[str, int | None] = {}
chosen_pmtu: Optional[int] = None chosen_pmtu: int | None = None
if targets: if targets:
log( log(
@@ -186,7 +190,7 @@ def run_automtu(args) -> int:
f"[automtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})" f"[automtu] Computed {args.wg_if} MTU: {wg_mtu} (overhead={args.wg_overhead}, min={args.wg_min})"
) )
wg_mtu_set: Optional[int] = None wg_mtu_set: int | None = None
wg_mtu_clamped = False wg_mtu_clamped = False
if args.set_wg_mtu is not None: if args.set_wg_mtu is not None:

View File

@@ -1,15 +1,13 @@
from __future__ import annotations from __future__ import annotations
import re import re
from typing import Optional
from .net import iface_exists, list_ifaces from .net import iface_exists, list_ifaces
_BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE) _BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE)
def _split_items(items: Optional[list[str]]) -> list[str]: def _split_items(items: list[str] | None) -> list[str]:
raw: list[str] = [] raw: list[str] = []
for item in items or []: for item in items or []:
raw.extend([x.strip() for x in item.split(",") if x.strip()]) raw.extend([x.strip() for x in item.split(",") if x.strip()])
@@ -18,7 +16,7 @@ def _split_items(items: Optional[list[str]]) -> list[str]:
def detect_docker_ifaces( def detect_docker_ifaces(
docker_if_args: Optional[list[str]], *, include_user_bridges: bool docker_if_args: list[str] | None, *, include_user_bridges: bool
) -> list[str]: ) -> list[str]:
""" """
Determine Docker-related interfaces to apply MTU to. Determine Docker-related interfaces to apply MTU to.

View File

@@ -5,7 +5,6 @@ import pathlib
import re import re
import subprocess import subprocess
import sys import sys
from typing import Optional
def _run(cmd: list[str]) -> str: def _run(cmd: list[str]) -> str:
@@ -14,27 +13,55 @@ def _run(cmd: list[str]) -> str:
).stdout.strip() ).stdout.strip()
SYSFS_NET = pathlib.Path("/sys/class/net")
def _ip_link_show(iface: str) -> str:
return _run(["ip", "link", "show", "dev", iface])
def iface_exists(iface: str) -> bool: def iface_exists(iface: str) -> bool:
return pathlib.Path(f"/sys/class/net/{iface}").exists() """
True if the interface exists. Does not follow the /sys/class/net symlink;
falls back to netlink when sysfs is unavailable.
"""
if os.path.lexists(SYSFS_NET / iface):
return True
return bool(_ip_link_show(iface))
def list_ifaces() -> list[str]: def list_ifaces() -> list[str]:
""" """
Return a sorted list of all network interfaces visible under /sys/class/net. Return a sorted list of all network interfaces, netlink as fallback.
Plain files in /sys/class/net (e.g. bonding_masters) are not interfaces.
""" """
base = pathlib.Path("/sys/class/net") try:
if not base.exists(): return sorted(
return [] p.name for p in SYSFS_NET.iterdir() if p.is_symlink() or p.is_dir()
names: list[str] = [] )
for p in base.iterdir(): except OSError:
if p.is_dir(): out = _run(["ip", "-o", "link", "show"])
names.append(p.name) return sorted(re.findall(r"^\d+:\s+([^:@\s]+)", out, flags=re.MULTILINE))
names.sort()
return names
def read_iface_mtu(iface: str) -> int: def read_iface_mtu(iface: str) -> int:
return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip()) """
Read the interface MTU from sysfs, falling back to netlink.
Raises RuntimeError if neither source reports an MTU.
"""
try:
return int((SYSFS_NET / iface / "mtu").read_text().strip())
except (OSError, ValueError):
pass
m = re.search(r"\bmtu\s+(\d+)\b", _ip_link_show(iface))
if not m:
raise RuntimeError(
f"Could not read MTU of {iface}: "
f"{SYSFS_NET / iface / 'mtu'} is unreadable and "
f"'ip link show dev {iface}' reported no MTU."
)
return int(m.group(1))
def set_iface_mtu(iface: str, mtu: int, dry: bool) -> None: def set_iface_mtu(iface: str, mtu: int, dry: bool) -> None:
@@ -53,7 +80,7 @@ def require_root(*, dry: bool, needs_root: bool) -> None:
raise SystemExit(1) raise SystemExit(1)
def detect_egress_iface(ignore_vpn: bool = True) -> Optional[str]: def detect_egress_iface(ignore_vpn: bool = True) -> str | None:
devs: list[str] = [] devs: list[str] = []
for cmd in ( for cmd in (
["ip", "-4", "route", "show", "default"], ["ip", "-4", "route", "show", "default"],

View File

@@ -4,19 +4,18 @@ from __future__ import annotations
import json import json
import sys import sys
from dataclasses import dataclass from dataclasses import dataclass
from typing import Optional
@dataclass(frozen=True) @dataclass(frozen=True)
class OutputMode: class OutputMode:
print_mtu: Optional[str] # "egress" | "effective" | "wg" | None print_mtu: str | None # "egress" | "effective" | "wg" | None
print_json: bool print_json: bool
@property @property
def machine(self) -> bool: def machine(self) -> bool:
return bool(self.print_mtu or self.print_json) return bool(self.print_mtu or self.print_json)
def validate(self) -> Optional[str]: def validate(self) -> str | None:
if self.print_mtu and self.print_json: if self.print_mtu and self.print_json:
return "--print-mtu and --print-json are mutually exclusive." return "--print-mtu and --print-json are mutually exclusive."
return None return None
@@ -67,24 +66,24 @@ def emit_json(
egress_iface: str, egress_iface: str,
base_mtu: int, base_mtu: int,
effective_mtu: int, effective_mtu: int,
egress_forced_mtu: Optional[int], egress_forced_mtu: int | None,
egress_applied: bool, egress_applied: bool,
pmtu_targets: list[str], pmtu_targets: list[str],
pmtu_auto_targets_added: list[str], pmtu_auto_targets_added: list[str],
pmtu_policy: str, pmtu_policy: str,
pmtu_chosen: Optional[int], pmtu_chosen: int | None,
pmtu_results: dict[str, Optional[int]], pmtu_results: dict[str, int | None],
wg_iface: str, wg_iface: str,
wg_mtu: int, wg_mtu: int,
wg_overhead: int, wg_overhead: int,
wg_min: int, wg_min: int,
wg_set_mtu: Optional[int], wg_set_mtu: int | None,
wg_clamped: bool, wg_clamped: bool,
wg_present: bool, wg_present: bool,
wg_active: bool, wg_active: bool,
wg_applied: bool, wg_applied: bool,
docker_ifaces: Optional[list[str]] = None, docker_ifaces: list[str] | None = None,
docker_applied: Optional[list[str]] = None, docker_applied: list[str] | None = None,
dry_run: bool, dry_run: bool,
) -> bool: ) -> bool:
""" """

View File

@@ -4,14 +4,12 @@ import shlex
import shutil import shutil
import subprocess import subprocess
from pathlib import Path from pathlib import Path
from typing import List
_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu.service") _SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu.service")
_DOCKER_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu-docker.service") _DOCKER_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu-docker.service")
def _strip_persist_args(argv: List[str]) -> List[str]: def _strip_persist_args(argv: list[str]) -> list[str]:
""" """
Remove persistence-only arguments from argv: Remove persistence-only arguments from argv:
- --persist systemd|docker - --persist systemd|docker
@@ -51,7 +49,7 @@ def _resolve_exec(argv0: str) -> str:
return argv0 return argv0
def _needs_docker_ordering(filtered_argv: List[str]) -> bool: def _needs_docker_ordering(filtered_argv: list[str]) -> bool:
""" """
Heuristic: If we apply docker MTU (directly or via --apply-all), order after docker.service. Heuristic: If we apply docker MTU (directly or via --apply-all), order after docker.service.
""" """
@@ -115,7 +113,7 @@ def _uninstall_unit(unit_path: Path, *, dry: bool) -> None:
print(f"[automtu] Uninstalled systemd service: {unit_path.name}") print(f"[automtu] Uninstalled systemd service: {unit_path.name}")
def persist_systemd(argv: List[str], *, dry: bool) -> None: def persist_systemd(argv: list[str], *, dry: bool) -> None:
""" """
Install a systemd oneshot service that re-runs automtu with the same arguments. Install a systemd oneshot service that re-runs automtu with the same arguments.
Adds docker ordering automatically if docker MTU is applied. Adds docker ordering automatically if docker MTU is applied.
@@ -142,7 +140,7 @@ def uninstall_systemd(*, dry: bool) -> None:
_uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry) _uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry)
def persist_docker(argv: List[str], *, dry: bool) -> None: def persist_docker(argv: list[str], *, dry: bool) -> None:
""" """
Docker-focused persistence backend: Docker-focused persistence backend:
always orders after docker.service (even if args don't include docker flags), always orders after docker.service (even if args don't include docker flags),

View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import ipaddress import ipaddress
import subprocess import subprocess
from typing import Optional
def _is_ipv6(target: str) -> bool: def _is_ipv6(target: str) -> bool:
@@ -14,7 +13,7 @@ def _is_ipv6(target: str) -> bool:
def _rc(cmd: list[str]) -> int: def _rc(cmd: list[str]) -> int:
return subprocess.run( return subprocess.run(
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL cmd, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
).returncode ).returncode
@@ -28,7 +27,7 @@ def _ping_ok(payload: int, target: str, timeout_s: float) -> bool:
"-s", "-s",
str(payload), str(payload),
"-W", "-W",
str(max(1, int(round(timeout_s)))), str(max(1, round(timeout_s))),
] ]
if _is_ipv6(target): if _is_ipv6(target):
cmd.insert(1, "-6") cmd.insert(1, "-6")
@@ -37,7 +36,7 @@ def _ping_ok(payload: int, target: str, timeout_s: float) -> bool:
def probe_pmtu( def probe_pmtu(
target: str, lo_payload: int = 1200, hi_payload: int = 1472, timeout: float = 1.0 target: str, lo_payload: int = 1200, hi_payload: int = 1472, timeout: float = 1.0
) -> Optional[int]: ) -> int | None:
hdr = 48 if _is_ipv6(target) else 28 hdr = 48 if _is_ipv6(target) else 28
if not _ping_ok(lo_payload, target, timeout): if not _ping_ok(lo_payload, target, timeout):

View File

@@ -2,7 +2,6 @@ from __future__ import annotations
import re import re
import subprocess import subprocess
from typing import List
from .net import iface_exists from .net import iface_exists
@@ -15,7 +14,7 @@ def _run(cmd: list[str]) -> str:
def _rc(cmd: list[str]) -> int: def _rc(cmd: list[str]) -> int:
return subprocess.run( return subprocess.run(
cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL cmd, check=False, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL
).returncode ).returncode
@@ -23,7 +22,7 @@ def wg_is_active(wg_if: str) -> bool:
return iface_exists(wg_if) and _rc(["wg", "show", wg_if]) == 0 return iface_exists(wg_if) and _rc(["wg", "show", wg_if]) == 0
def wg_peer_endpoints(wg_if: str) -> List[str]: def wg_peer_endpoints(wg_if: str) -> list[str]:
targets: list[str] = [] targets: list[str] = []
out = _run(["wg", "show", wg_if, "endpoints"]) out = _run(["wg", "show", wg_if, "endpoints"])

View File

@@ -1,12 +1,46 @@
import io import io
import pathlib
import tempfile
import unittest import unittest
from contextlib import redirect_stdout from contextlib import redirect_stdout
from types import SimpleNamespace from types import SimpleNamespace
from unittest.mock import patch from unittest.mock import patch
from automtu import net
from automtu.core import run_automtu 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): class TestCore(unittest.TestCase):
def test_run_automtu_happy_path_all_mocked(self) -> None: def test_run_automtu_happy_path_all_mocked(self) -> None:
args = SimpleNamespace( args = SimpleNamespace(
@@ -164,5 +198,53 @@ class TestCore(unittest.TestCase):
mock_set.assert_not_called() 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__": if __name__ == "__main__":
unittest.main(verbosity=2) unittest.main(verbosity=2)

View File

@@ -1,7 +1,7 @@
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
import automtu.docker as docker from automtu import docker
class TestDocker(unittest.TestCase): class TestDocker(unittest.TestCase):

View File

@@ -1,8 +1,120 @@
import pathlib
import tempfile
import unittest import unittest
from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
import automtu.net as net 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): class TestNet(unittest.TestCase):
@@ -46,18 +158,10 @@ class TestNet(unittest.TestCase):
self.assertFalse(net.default_route_uses_iface("wg0")) self.assertFalse(net.default_route_uses_iface("wg0"))
def test_list_ifaces_returns_sorted_names(self) -> None: def test_list_ifaces_returns_sorted_names(self) -> None:
fake = [ with tempfile.TemporaryDirectory() as tmp:
Path("/sys/class/net/eth0"), netdir = _sysbox_sysfs(pathlib.Path(tmp), ("wg0", "eth0", "lo"))
Path("/sys/class/net/lo"), with patch.object(net, "SYSFS_NET", netdir):
Path("/sys/class/net/wg0"), self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])
]
with (
patch("automtu.net.pathlib.Path.exists", return_value=True),
patch("automtu.net.pathlib.Path.iterdir", return_value=fake),
patch.object(Path, "is_dir", return_value=True),
):
self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])
if __name__ == "__main__": if __name__ == "__main__":

View File

@@ -1,9 +1,9 @@
import io import io
import json import json
import unittest import unittest
from contextlib import redirect_stdout, redirect_stderr from contextlib import redirect_stderr, redirect_stdout
from automtu.output import OutputMode, emit_json, emit_single_number, Logger from automtu.output import Logger, OutputMode, emit_json, emit_single_number
class TestOutput(unittest.TestCase): class TestOutput(unittest.TestCase):

View File

@@ -4,7 +4,7 @@ from contextlib import redirect_stdout
from pathlib import Path from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
import automtu.persist as persist from automtu import persist
class TestPersist(unittest.TestCase): class TestPersist(unittest.TestCase):

View File

@@ -1,7 +1,7 @@
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
import automtu.pmtu as pmtu from automtu import pmtu
class TestPmtu(unittest.TestCase): class TestPmtu(unittest.TestCase):

View File

@@ -1,7 +1,7 @@
import unittest import unittest
from unittest.mock import patch from unittest.mock import patch
import automtu.wg as wg from automtu import wg
class TestWg(unittest.TestCase): class TestWg(unittest.TestCase):