4 Commits

18 changed files with 624 additions and 30 deletions

5
.gitignore vendored
View File

@@ -1,2 +1,5 @@
*.pyc *.pyc
__pycache__ __pycache__
dist/
build/*
*.egg-info

View File

@@ -1,3 +1,10 @@
## [1.1.0] - 2026-01-23
* * Added persistent MTU configuration via systemd (install & uninstall)
* Added Docker bridge MTU support and Docker-ordered persistence backend
* Added repository mirrors and ignored build artifacts
## [1.0.0] - 2026-01-21 ## [1.0.0] - 2026-01-21
* 🥳 Official Release * 🥳 Official Release

4
MIRRORS Normal file
View File

@@ -0,0 +1,4 @@
git@github.com:kevinveenbirkenbach/wg-mtu-auto.git
ssh://git@git.veen.world:2201/kevinveenbirkenbach/automtu.git
ssh://git@code.infinito.nexus:2201/kevinveenbirkenbach/automtu.git
https://pypi.org/project/automtu/

View File

@@ -10,3 +10,6 @@ help:
test: test:
@PYTHONPATH="$(CURDIR)/src" "$(PYTHON)" -m unittest discover -s tests/unit -p "test_*.py" -v @PYTHONPATH="$(CURDIR)/src" "$(PYTHON)" -m unittest discover -s tests/unit -p "test_*.py" -v
install:
sudo pip install -e . --upgrade --break-system-packages

View File

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "automtu" name = "automtu"
version = "1.0.0" version = "1.1.0"
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

@@ -1,2 +0,0 @@
__all__ = ["__version__"]
__version__ = "0.1.0"

View File

@@ -7,7 +7,7 @@ import os
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
ap = argparse.ArgumentParser( ap = argparse.ArgumentParser(
prog="automtu", prog="automtu",
description="Probe Path MTU and compute/apply MTU for egress and/or WireGuard.", description="Probe Path MTU and compute/apply MTU for egress and/or WireGuard (and Docker bridges).",
) )
ap.add_argument( ap.add_argument(
@@ -49,6 +49,7 @@ def build_parser() -> argparse.ArgumentParser:
help="Aggregate PMTU across targets (default: min).", help="Aggregate PMTU across targets (default: min).",
) )
# --- Apply flags ---
ap.add_argument( ap.add_argument(
"--apply-egress-mtu", "--apply-egress-mtu",
action="store_true", action="store_true",
@@ -57,7 +58,18 @@ def build_parser() -> argparse.ArgumentParser:
ap.add_argument( ap.add_argument(
"--apply-wg-mtu", action="store_true", help="Apply MTU to WireGuard interface." "--apply-wg-mtu", action="store_true", help="Apply MTU to WireGuard interface."
) )
ap.add_argument(
"--apply-docker-mtu",
action="store_true",
help="Apply effective MTU to Docker bridges (docker0 and br-*).",
)
ap.add_argument(
"--apply-all",
action="store_true",
help="Apply MTU to egress + WireGuard + Docker bridges (implies apply flags).",
)
# --- WireGuard ---
ap.add_argument( ap.add_argument(
"--wg-if", "--wg-if",
default=os.environ.get("WG_IF", "wg0"), default=os.environ.get("WG_IF", "wg0"),
@@ -82,6 +94,19 @@ def build_parser() -> argparse.ArgumentParser:
) )
ap.add_argument("--set-wg-mtu", type=int, help="Force MTU for WireGuard interface.") ap.add_argument("--set-wg-mtu", type=int, help="Force MTU for WireGuard interface.")
# --- Docker ---
ap.add_argument(
"--docker-if",
action="append",
help="Explicit Docker interface(s) for MTU apply (repeatable or comma-separated). If omitted, auto-detect docker0 and br-*.",
)
ap.add_argument(
"--docker-no-user-bridges",
action="store_true",
help="Only apply to docker0, do not include br-* user bridges.",
)
# --- Force egress ---
ap.add_argument( ap.add_argument(
"--force-egress-mtu", "--force-egress-mtu",
type=int, type=int,
@@ -91,6 +116,18 @@ def build_parser() -> argparse.ArgumentParser:
"--dry-run", action="store_true", help="Show actions without applying changes." "--dry-run", action="store_true", help="Show actions without applying changes."
) )
# --- Persistence ---
ap.add_argument(
"--persist",
choices=["systemd", "docker"],
help="Persist MTU configuration across reboots (supported: systemd, docker).",
)
ap.add_argument(
"--uninstall",
action="store_true",
help="Uninstall persistence backend (requires --persist).",
)
# --- Machine-readable output modes --- # --- Machine-readable output modes ---
ap.add_argument( ap.add_argument(
"--print-mtu", "--print-mtu",

View File

@@ -5,6 +5,7 @@ import sys
from dataclasses import dataclass from dataclasses import dataclass
from typing import Iterable, Optional from typing import Iterable, Optional
from .docker import detect_docker_ifaces
from .net import ( from .net import (
default_route_uses_iface, default_route_uses_iface,
detect_egress_iface, detect_egress_iface,
@@ -46,6 +47,12 @@ def _choose(values: Iterable[int], policy: str) -> int:
def run_automtu(args) -> int: def run_automtu(args) -> int:
# Expand apply-all -> set apply flags
if getattr(args, "apply_all", False):
args.apply_egress_mtu = True
args.apply_wg_mtu = True
args.apply_docker_mtu = True
mode = OutputMode( mode = OutputMode(
print_mtu=getattr(args, "print_mtu", None), print_mtu=getattr(args, "print_mtu", None),
print_json=bool(getattr(args, "print_json", False)), print_json=bool(getattr(args, "print_json", False)),
@@ -57,14 +64,32 @@ def run_automtu(args) -> int:
log = Logger(mode.machine).log log = Logger(mode.machine).log
# Root is only needed if we actually change something (without --dry-run).
needs_root = bool( needs_root = bool(
args.apply_egress_mtu getattr(args, "apply_egress_mtu", False)
or args.apply_wg_mtu or getattr(args, "apply_wg_mtu", False)
or (args.force_egress_mtu is not None) or getattr(args, "apply_docker_mtu", False)
or (getattr(args, "force_egress_mtu", None) is not None)
or (getattr(args, "persist", None) is not None)
) )
require_root(dry=args.dry_run, needs_root=needs_root) require_root(dry=args.dry_run, needs_root=needs_root)
# Persistence mode: install/uninstall persistence mechanism and exit.
if getattr(args, "persist", None):
if args.persist == "systemd":
from .persist import persist_systemd, uninstall_systemd
if getattr(args, "uninstall", False):
uninstall_systemd(dry=args.dry_run)
return 0
persist_systemd(sys.argv, dry=args.dry_run)
return 0
print(
f"[automtu][ERROR] Unknown persist backend: {args.persist}", file=sys.stderr
)
return 4
egress = args.egress_if or detect_egress_iface(ignore_vpn=not args.prefer_wg_egress) egress = args.egress_if or detect_egress_iface(ignore_vpn=not args.prefer_wg_egress)
if not egress: if not egress:
print( print(
@@ -192,6 +217,29 @@ def run_automtu(args) -> int:
else: else:
log("[automtu] INFO: Not applying WireGuard MTU (use --apply-wg-mtu).") log("[automtu] INFO: Not applying WireGuard MTU (use --apply-wg-mtu).")
# Apply Docker MTU (optional)
docker_ifaces = detect_docker_ifaces(
getattr(args, "docker_if", None),
include_user_bridges=not bool(getattr(args, "docker_no_user_bridges", False)),
)
docker_applied: list[str] = []
if args.apply_docker_mtu:
if not docker_ifaces:
log("[automtu] INFO: No Docker interfaces detected for MTU apply.")
else:
log(
f"[automtu] Applying effective MTU {effective_mtu} to Docker ifaces: {', '.join(docker_ifaces)}"
)
for d in docker_ifaces:
if iface_exists(d):
set_iface_mtu(d, effective_mtu, args.dry_run)
docker_applied.append(d)
else:
log(
"[automtu] INFO: Not applying Docker MTU (use --apply-docker-mtu or --apply-all)."
)
# Machine-readable outputs # Machine-readable outputs
if emit_single_number( if emit_single_number(
mode, base_mtu=base_mtu, effective_mtu=effective_mtu, wg_mtu=wg_mtu mode, base_mtu=base_mtu, effective_mtu=effective_mtu, wg_mtu=wg_mtu
@@ -219,6 +267,8 @@ def run_automtu(args) -> int:
wg_present=wg_present, wg_present=wg_present,
wg_active=wg_active, wg_active=wg_active,
wg_applied=wg_applied, wg_applied=wg_applied,
docker_ifaces=docker_ifaces,
docker_applied=docker_applied,
dry_run=bool(args.dry_run), dry_run=bool(args.dry_run),
): ):
return 0 return 0

48
src/automtu/docker.py Normal file
View File

@@ -0,0 +1,48 @@
from __future__ import annotations
import re
from typing import Optional
from .net import iface_exists, list_ifaces
_BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE)
def _split_items(items: Optional[list[str]]) -> list[str]:
raw: list[str] = []
for item in items or []:
raw.extend([x.strip() for x in item.split(",") if x.strip()])
# de-dup while preserving order
return list(dict.fromkeys(raw))
def detect_docker_ifaces(
docker_if_args: Optional[list[str]], *, include_user_bridges: bool
) -> list[str]:
"""
Determine Docker-related interfaces to apply MTU to.
- If docker_if_args is given (repeatable / comma-separated), use those names (deduped).
- Otherwise auto-detect:
- docker0 (if exists)
- br-* user bridges (if include_user_bridges=True)
"""
explicit = _split_items(docker_if_args)
if explicit:
# keep only real interfaces; silently drop unknown names
return [i for i in explicit if iface_exists(i)]
found: list[str] = []
# docker0 is the classic default bridge
if iface_exists("docker0"):
found.append("docker0")
if include_user_bridges:
for name in list_ifaces():
if _BRIDGE_RE.match(name):
found.append(name)
# de-dup (defensive)
return list(dict.fromkeys(found))

View File

@@ -18,6 +18,21 @@ def iface_exists(iface: str) -> bool:
return pathlib.Path(f"/sys/class/net/{iface}").exists() return pathlib.Path(f"/sys/class/net/{iface}").exists()
def list_ifaces() -> list[str]:
"""
Return a sorted list of all network interfaces visible under /sys/class/net.
"""
base = pathlib.Path("/sys/class/net")
if not base.exists():
return []
names: list[str] = []
for p in base.iterdir():
if p.is_dir():
names.append(p.name)
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()) return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip())

View File

@@ -1,3 +1,4 @@
# src/automtu/output.py
from __future__ import annotations from __future__ import annotations
import json import json
@@ -27,7 +28,7 @@ class Logger:
""" """
def __init__(self, machine_mode: bool) -> None: def __init__(self, machine_mode: bool) -> None:
self._machine = machine_mode self._machine = bool(machine_mode)
def log(self, msg: str) -> None: def log(self, msg: str) -> None:
if self._machine: if self._machine:
@@ -56,7 +57,6 @@ def emit_single_number(
print(int(wg_mtu)) print(int(wg_mtu))
return True return True
# Should never happen due to argparse choices
print("[automtu][ERROR] Invalid --print-mtu value.", file=sys.stderr) print("[automtu][ERROR] Invalid --print-mtu value.", file=sys.stderr)
raise SystemExit(4) raise SystemExit(4)
@@ -83,6 +83,8 @@ def emit_json(
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_applied: Optional[list[str]] = None,
dry_run: bool, dry_run: bool,
) -> bool: ) -> bool:
""" """
@@ -91,6 +93,9 @@ def emit_json(
if not mode.print_json: if not mode.print_json:
return False return False
docker_ifaces = list(docker_ifaces or [])
docker_applied = list(docker_applied or [])
payload = { payload = {
"egress": { "egress": {
"iface": egress_iface, "iface": egress_iface,
@@ -121,6 +126,10 @@ def emit_json(
"active": bool(wg_active), "active": bool(wg_active),
"applied": bool(wg_applied), "applied": bool(wg_applied),
}, },
"docker": {
"ifaces": docker_ifaces,
"applied": docker_applied,
},
"dry_run": bool(dry_run), "dry_run": bool(dry_run),
} }

170
src/automtu/persist.py Normal file
View File

@@ -0,0 +1,170 @@
from __future__ import annotations
import shlex
import shutil
import subprocess
from pathlib import Path
from typing import List
_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu.service")
_DOCKER_SYSTEMD_UNIT_PATH = Path("/etc/systemd/system/automtu-docker.service")
def _strip_persist_args(argv: List[str]) -> List[str]:
"""
Remove persistence-only arguments from argv:
- --persist systemd|docker
- --persist=systemd|docker
- --uninstall
Keeps all other args as-is.
"""
out: list[str] = []
i = 0
while i < len(argv):
a = argv[i]
if a == "--persist":
i += 1
if i < len(argv) and not argv[i].startswith("-"):
i += 1
continue
if a.startswith("--persist="):
i += 1
continue
if a == "--uninstall":
i += 1
continue
out.append(a)
i += 1
return out
def _resolve_exec(argv0: str) -> str:
"""
Resolve the executable path for systemd ExecStart.
Prefer an absolute path (from PATH lookup). Fall back to argv0.
"""
resolved = shutil.which(argv0)
if resolved:
return resolved
return argv0
def _needs_docker_ordering(filtered_argv: List[str]) -> bool:
"""
Heuristic: If we apply docker MTU (directly or via --apply-all), order after docker.service.
"""
return ("--apply-docker-mtu" in filtered_argv) or ("--apply-all" in filtered_argv)
def _build_unit(execstart: str, *, docker_ordering: bool) -> str:
after_lines = ["network-online.target"]
wants_lines = ["network-online.target"]
if docker_ordering:
after_lines.append("docker.service")
wants_lines.append("docker.service")
after = " ".join(after_lines)
wants = " ".join(wants_lines)
return f"""\
[Unit]
Description=Auto MTU via automtu
After={after}
Wants={wants}
[Service]
Type=oneshot
ExecStart={execstart}
[Install]
WantedBy=multi-user.target
"""
def _install_unit(unit_path: Path, unit_text: str, *, dry: bool) -> None:
if dry:
print(f"[automtu] DRY-RUN: would write systemd unit to {unit_path}")
print(unit_text.rstrip())
print("[automtu] DRY-RUN: would run: systemctl daemon-reload")
print(f"[automtu] DRY-RUN: would run: systemctl enable {unit_path.name}")
return
unit_path.write_text(unit_text)
subprocess.run(["systemctl", "daemon-reload"], check=True)
subprocess.run(["systemctl", "enable", unit_path.name], check=True)
print(f"[automtu] Installed and enabled systemd service: {unit_path.name}")
print(
f"[automtu] Tip: run 'systemctl start {unit_path.name}' to apply immediately."
)
def _uninstall_unit(unit_path: Path, *, dry: bool) -> None:
if dry:
print(f"[automtu] DRY-RUN: would run: systemctl disable {unit_path.name}")
print(f"[automtu] DRY-RUN: would remove: {unit_path} (if exists)")
print("[automtu] DRY-RUN: would run: systemctl daemon-reload")
return
subprocess.run(["systemctl", "disable", unit_path.name], check=True)
if unit_path.exists():
unit_path.unlink()
subprocess.run(["systemctl", "daemon-reload"], check=True)
print(f"[automtu] Uninstalled systemd service: {unit_path.name}")
def persist_systemd(argv: List[str], *, dry: bool) -> None:
"""
Install a systemd oneshot service that re-runs automtu with the same arguments.
Adds docker ordering automatically if docker MTU is applied.
"""
if not argv:
raise ValueError("argv must not be empty")
filtered = _strip_persist_args(argv[:])
if not filtered:
raise ValueError("argv filtered to empty; cannot persist")
exe = _resolve_exec(filtered[0])
args = [exe, *filtered[1:]]
execstart = shlex.join(args)
unit = _build_unit(execstart, docker_ordering=_needs_docker_ordering(filtered))
_install_unit(_SYSTEMD_UNIT_PATH, unit, dry=dry)
def uninstall_systemd(*, dry: bool) -> None:
"""
Uninstall the base systemd persistence backend.
"""
_uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry)
def persist_docker(argv: List[str], *, dry: bool) -> None:
"""
Docker-focused persistence backend:
always orders after docker.service (even if args don't include docker flags),
because the user chose it explicitly.
"""
if not argv:
raise ValueError("argv must not be empty")
filtered = _strip_persist_args(argv[:])
if not filtered:
raise ValueError("argv filtered to empty; cannot persist")
exe = _resolve_exec(filtered[0])
args = [exe, *filtered[1:]]
execstart = shlex.join(args)
unit = _build_unit(execstart, docker_ordering=True)
_install_unit(_DOCKER_SYSTEMD_UNIT_PATH, unit, dry=dry)
def uninstall_docker(*, dry: bool) -> None:
"""
Uninstall the docker-ordered systemd backend.
"""
_uninstall_unit(_DOCKER_SYSTEMD_UNIT_PATH, dry=dry)

View File

@@ -1,14 +0,0 @@
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)

View File

@@ -23,8 +23,7 @@ class TestCli(unittest.TestCase):
"1472", "1472",
"--pmtu-policy", "--pmtu-policy",
"median", "median",
"--apply-egress-mtu", "--apply-all",
"--apply-wg-mtu",
"--wg-if", "--wg-if",
"wg0", "wg0",
"--wg-overhead", "--wg-overhead",
@@ -36,6 +35,8 @@ class TestCli(unittest.TestCase):
"1372", "1372",
"--force-egress-mtu", "--force-egress-mtu",
"1452", "1452",
"--docker-if",
"docker0,br-123",
"--dry-run", "--dry-run",
"--print-mtu", "--print-mtu",
"wg", "wg",
@@ -49,14 +50,22 @@ class TestCli(unittest.TestCase):
self.assertEqual(args.pmtu_min_payload, 1200) self.assertEqual(args.pmtu_min_payload, 1200)
self.assertEqual(args.pmtu_max_payload, 1472) self.assertEqual(args.pmtu_max_payload, 1472)
self.assertEqual(args.pmtu_policy, "median") self.assertEqual(args.pmtu_policy, "median")
self.assertTrue(args.apply_egress_mtu)
self.assertTrue(args.apply_wg_mtu) self.assertTrue(args.apply_all)
self.assertFalse(args.apply_egress_mtu) # apply_all expands in core()
self.assertFalse(args.apply_wg_mtu)
self.assertFalse(args.apply_docker_mtu)
self.assertEqual(args.wg_if, "wg0") self.assertEqual(args.wg_if, "wg0")
self.assertEqual(args.wg_overhead, 80) self.assertEqual(args.wg_overhead, 80)
self.assertEqual(args.wg_min, 1280) self.assertEqual(args.wg_min, 1280)
self.assertTrue(args.auto_pmtu_from_wg) self.assertTrue(args.auto_pmtu_from_wg)
self.assertEqual(args.set_wg_mtu, 1372) self.assertEqual(args.set_wg_mtu, 1372)
self.assertEqual(args.force_egress_mtu, 1452) self.assertEqual(args.force_egress_mtu, 1452)
self.assertEqual(args.docker_if, ["docker0,br-123"])
self.assertFalse(args.docker_no_user_bridges)
self.assertTrue(args.dry_run) self.assertTrue(args.dry_run)
self.assertEqual(args.print_mtu, "wg") self.assertEqual(args.print_mtu, "wg")
self.assertFalse(args.print_json) self.assertFalse(args.print_json)

View File

@@ -22,10 +22,18 @@ class TestCore(unittest.TestCase):
pmtu_policy="min", pmtu_policy="min",
apply_egress_mtu=True, apply_egress_mtu=True,
apply_wg_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_if="wg0",
wg_overhead=80, wg_overhead=80,
wg_min=1280, wg_min=1280,
set_wg_mtu=None, 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 # PMTU probes: 1452 and 1500 -> min policy => 1452, effective=min(base(1500),1452)=1452
@@ -40,6 +48,7 @@ class TestCore(unittest.TestCase):
patch("automtu.core.wg_is_active", return_value=False), patch("automtu.core.wg_is_active", return_value=False),
patch("automtu.core.wg_peer_endpoints", return_value=[]), patch("automtu.core.wg_peer_endpoints", return_value=[]),
patch("automtu.core.default_route_uses_iface", return_value=False), patch("automtu.core.default_route_uses_iface", return_value=False),
patch("automtu.core.detect_docker_ifaces", return_value=[]),
): ):
buf = io.StringIO() buf = io.StringIO()
with redirect_stdout(buf): with redirect_stdout(buf):
@@ -52,10 +61,66 @@ class TestCore(unittest.TestCase):
self.assertIn("Selected Path MTU (policy=min): 1452", s) self.assertIn("Selected Path MTU (policy=min): 1452", s)
self.assertIn("Computed wg0 MTU: 1372", 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("eth0", 1452, True)
mock_set.assert_any_call("wg0", 1372, 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: def test_run_automtu_does_not_apply_wg_without_flag(self) -> None:
args = SimpleNamespace( args = SimpleNamespace(
dry_run=True, dry_run=True,
@@ -70,10 +135,18 @@ class TestCore(unittest.TestCase):
pmtu_policy="min", pmtu_policy="min",
apply_egress_mtu=False, apply_egress_mtu=False,
apply_wg_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_if="wg0",
wg_overhead=80, wg_overhead=80,
wg_min=1280, wg_min=1280,
set_wg_mtu=None, set_wg_mtu=None,
persist=None,
uninstall=False,
print_mtu=None,
print_json=False,
) )
with ( with (
@@ -81,6 +154,7 @@ class TestCore(unittest.TestCase):
patch("automtu.core.iface_exists", return_value=True), patch("automtu.core.iface_exists", return_value=True),
patch("automtu.core.read_iface_mtu", return_value=1500), patch("automtu.core.read_iface_mtu", return_value=1500),
patch("automtu.core.set_iface_mtu") as mock_set, patch("automtu.core.set_iface_mtu") as mock_set,
patch("automtu.core.detect_docker_ifaces", return_value=[]),
): ):
buf = io.StringIO() buf = io.StringIO()
with redirect_stdout(buf): with redirect_stdout(buf):

49
tests/unit/test_docker.py Normal file
View File

@@ -0,0 +1,49 @@
import unittest
from unittest.mock import patch
import automtu.docker as docker
class TestDocker(unittest.TestCase):
def test_detect_docker_ifaces_explicit_dedup_and_drop_unknown(self) -> None:
# explicit args are repeatable and/or comma-separated
args = ["docker0,br-abc,unknown0", "br-abc,docker0"]
def fake_iface_exists(name: str) -> bool:
return name in {"docker0", "br-abc"} # unknown0 does not exist
with patch("automtu.docker.iface_exists", side_effect=fake_iface_exists):
got = docker.detect_docker_ifaces(args, include_user_bridges=True)
# preserve order, de-dup, drop unknown
self.assertEqual(got, ["docker0", "br-abc"])
def test_detect_docker_ifaces_auto_detect_docker0_only(self) -> None:
# no explicit args -> auto detect
def fake_iface_exists(name: str) -> bool:
return name == "docker0"
with patch("automtu.docker.iface_exists", side_effect=fake_iface_exists):
got = docker.detect_docker_ifaces(None, include_user_bridges=False)
self.assertEqual(got, ["docker0"])
def test_detect_docker_ifaces_auto_detect_includes_user_bridges(self) -> None:
# docker0 exists + br-* exists in list_ifaces -> included if include_user_bridges=True
def fake_iface_exists(name: str) -> bool:
return name in {"docker0", "br-abc"} # br-abc exists, br-nope does not
with (
patch("automtu.docker.iface_exists", side_effect=fake_iface_exists),
patch(
"automtu.docker.list_ifaces",
return_value=["lo", "eth0", "br-abc", "br-nope", "wg0"],
),
):
got = docker.detect_docker_ifaces(None, include_user_bridges=True)
self.assertEqual(got, ["docker0", "br-abc"])
if __name__ == "__main__":
unittest.main(verbosity=2)

View File

@@ -1,4 +1,5 @@
import unittest import unittest
from pathlib import Path
from unittest.mock import patch from unittest.mock import patch
import automtu.net as net import automtu.net as net
@@ -44,6 +45,20 @@ class TestNet(unittest.TestCase):
self.assertTrue(net.default_route_uses_iface("eth0")) self.assertTrue(net.default_route_uses_iface("eth0"))
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:
fake = [
Path("/sys/class/net/eth0"),
Path("/sys/class/net/lo"),
Path("/sys/class/net/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__":
unittest.main(verbosity=2) unittest.main(verbosity=2)

117
tests/unit/test_persist.py Normal file
View File

@@ -0,0 +1,117 @@
import io
import unittest
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import patch
import automtu.persist as persist
class TestPersist(unittest.TestCase):
def test_strip_persist_args_removes_flag_and_value(self) -> None:
argv = ["automtu", "--apply-wg-mtu", "--persist", "systemd"]
self.assertEqual(
persist._strip_persist_args(argv),
["automtu", "--apply-wg-mtu"],
)
def test_strip_persist_args_removes_equals_form(self) -> None:
argv = ["automtu", "--apply-wg-mtu", "--persist=systemd"]
self.assertEqual(
persist._strip_persist_args(argv),
["automtu", "--apply-wg-mtu"],
)
def test_strip_persist_args_removes_uninstall(self) -> None:
argv = ["automtu", "--persist", "docker", "--uninstall", "--apply-wg-mtu"]
self.assertEqual(
persist._strip_persist_args(argv),
["automtu", "--apply-wg-mtu"],
)
def test_persist_systemd_dry_run_prints_unit(self) -> None:
argv = [
"automtu",
"--auto-pmtu-from-wg",
"--apply-wg-mtu",
"--persist",
"systemd",
]
with (
patch("automtu.persist.shutil.which", return_value="/usr/bin/automtu"),
patch("automtu.persist._SYSTEMD_UNIT_PATH", Path("/tmp/automtu.service")),
):
out = io.StringIO()
with redirect_stdout(out):
persist.persist_systemd(argv, dry=True)
s = out.getvalue()
self.assertIn("DRY-RUN", s)
self.assertIn(
"ExecStart=/usr/bin/automtu --auto-pmtu-from-wg --apply-wg-mtu", s
)
def test_persist_systemd_adds_docker_ordering_if_apply_docker_mtu_present(
self,
) -> None:
argv = ["automtu", "--apply-docker-mtu", "--persist", "systemd"]
with (
patch("automtu.persist.shutil.which", return_value="/usr/bin/automtu"),
patch("automtu.persist._SYSTEMD_UNIT_PATH", Path("/tmp/automtu.service")),
):
out = io.StringIO()
with redirect_stdout(out):
persist.persist_systemd(argv, dry=True)
s = out.getvalue()
self.assertIn("After=network-online.target docker.service", s)
self.assertIn("Wants=network-online.target docker.service", s)
def test_persist_docker_dry_run_prints_unit_with_docker_ordering(self) -> None:
argv = ["automtu", "--dry-run", "--persist", "docker"]
with (
patch("automtu.persist.shutil.which", return_value="/usr/bin/automtu"),
patch(
"automtu.persist._DOCKER_SYSTEMD_UNIT_PATH",
Path("/tmp/automtu-docker.service"),
),
):
out = io.StringIO()
with redirect_stdout(out):
persist.persist_docker(argv, dry=True)
s = out.getvalue()
self.assertIn("DRY-RUN", s)
self.assertIn("After=network-online.target docker.service", s)
self.assertIn("Wants=network-online.target docker.service", s)
self.assertIn("ExecStart=/usr/bin/automtu --dry-run", s)
def test_uninstall_systemd_dry_run_prints_actions(self) -> None:
with patch("automtu.persist._SYSTEMD_UNIT_PATH", Path("/tmp/automtu.service")):
out = io.StringIO()
with redirect_stdout(out):
persist.uninstall_systemd(dry=True)
s = out.getvalue()
self.assertIn("DRY-RUN", s)
self.assertIn("systemctl disable", s)
def test_uninstall_docker_dry_run_prints_actions(self) -> None:
with patch(
"automtu.persist._DOCKER_SYSTEMD_UNIT_PATH",
Path("/tmp/automtu-docker.service"),
):
out = io.StringIO()
with redirect_stdout(out):
persist.uninstall_docker(dry=True)
s = out.getvalue()
self.assertIn("DRY-RUN", s)
self.assertIn("automtu-docker.service", s)
if __name__ == "__main__":
unittest.main(verbosity=2)