Compare commits
3 Commits
77da7a4acb
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| a0db5f797f | |||
| fc5dee4dc0 | |||
| f534d025d0 |
1
.gitignore
vendored
1
.gitignore
vendored
@@ -1,4 +1,5 @@
|
||||
*.pyc
|
||||
__pycache__
|
||||
dist/
|
||||
build/*
|
||||
*.egg-info
|
||||
@@ -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
|
||||
|
||||
* 🥳 Official Release
|
||||
|
||||
3
Makefile
3
Makefile
@@ -10,3 +10,6 @@ help:
|
||||
|
||||
test:
|
||||
@PYTHONPATH="$(CURDIR)/src" "$(PYTHON)" -m unittest discover -s tests/unit -p "test_*.py" -v
|
||||
|
||||
install:
|
||||
sudo pip install -e . --upgrade --break-system-packages
|
||||
|
||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "automtu"
|
||||
version = "1.0.0"
|
||||
version = "1.1.0"
|
||||
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.10"
|
||||
|
||||
@@ -1,2 +0,0 @@
|
||||
__all__ = ["__version__"]
|
||||
__version__ = "0.1.0"
|
||||
|
||||
@@ -7,7 +7,7 @@ import os
|
||||
def build_parser() -> argparse.ArgumentParser:
|
||||
ap = argparse.ArgumentParser(
|
||||
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(
|
||||
@@ -49,6 +49,7 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
help="Aggregate PMTU across targets (default: min).",
|
||||
)
|
||||
|
||||
# --- Apply flags ---
|
||||
ap.add_argument(
|
||||
"--apply-egress-mtu",
|
||||
action="store_true",
|
||||
@@ -57,7 +58,18 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
ap.add_argument(
|
||||
"--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(
|
||||
"--wg-if",
|
||||
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.")
|
||||
|
||||
# --- 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(
|
||||
"--force-egress-mtu",
|
||||
type=int,
|
||||
@@ -91,6 +116,18 @@ def build_parser() -> argparse.ArgumentParser:
|
||||
"--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 ---
|
||||
ap.add_argument(
|
||||
"--print-mtu",
|
||||
|
||||
@@ -5,6 +5,7 @@ import sys
|
||||
from dataclasses import dataclass
|
||||
from typing import Iterable, Optional
|
||||
|
||||
from .docker import detect_docker_ifaces
|
||||
from .net import (
|
||||
default_route_uses_iface,
|
||||
detect_egress_iface,
|
||||
@@ -46,6 +47,12 @@ def _choose(values: Iterable[int], policy: str) -> 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(
|
||||
print_mtu=getattr(args, "print_mtu", None),
|
||||
print_json=bool(getattr(args, "print_json", False)),
|
||||
@@ -57,14 +64,32 @@ def run_automtu(args) -> int:
|
||||
|
||||
log = Logger(mode.machine).log
|
||||
|
||||
# Root is only needed if we actually change something (without --dry-run).
|
||||
needs_root = bool(
|
||||
args.apply_egress_mtu
|
||||
or args.apply_wg_mtu
|
||||
or (args.force_egress_mtu is not None)
|
||||
getattr(args, "apply_egress_mtu", False)
|
||||
or getattr(args, "apply_wg_mtu", False)
|
||||
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)
|
||||
|
||||
# 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)
|
||||
if not egress:
|
||||
print(
|
||||
@@ -192,6 +217,29 @@ def run_automtu(args) -> int:
|
||||
else:
|
||||
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
|
||||
if emit_single_number(
|
||||
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_active=wg_active,
|
||||
wg_applied=wg_applied,
|
||||
docker_ifaces=docker_ifaces,
|
||||
docker_applied=docker_applied,
|
||||
dry_run=bool(args.dry_run),
|
||||
):
|
||||
return 0
|
||||
|
||||
48
src/automtu/docker.py
Normal file
48
src/automtu/docker.py
Normal 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))
|
||||
@@ -18,6 +18,21 @@ def iface_exists(iface: str) -> bool:
|
||||
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:
|
||||
return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip())
|
||||
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
# src/automtu/output.py
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
@@ -27,7 +28,7 @@ class Logger:
|
||||
"""
|
||||
|
||||
def __init__(self, machine_mode: bool) -> None:
|
||||
self._machine = machine_mode
|
||||
self._machine = bool(machine_mode)
|
||||
|
||||
def log(self, msg: str) -> None:
|
||||
if self._machine:
|
||||
@@ -56,7 +57,6 @@ def emit_single_number(
|
||||
print(int(wg_mtu))
|
||||
return True
|
||||
|
||||
# Should never happen due to argparse choices
|
||||
print("[automtu][ERROR] Invalid --print-mtu value.", file=sys.stderr)
|
||||
raise SystemExit(4)
|
||||
|
||||
@@ -83,6 +83,8 @@ def emit_json(
|
||||
wg_present: bool,
|
||||
wg_active: bool,
|
||||
wg_applied: bool,
|
||||
docker_ifaces: Optional[list[str]] = None,
|
||||
docker_applied: Optional[list[str]] = None,
|
||||
dry_run: bool,
|
||||
) -> bool:
|
||||
"""
|
||||
@@ -91,6 +93,9 @@ def emit_json(
|
||||
if not mode.print_json:
|
||||
return False
|
||||
|
||||
docker_ifaces = list(docker_ifaces or [])
|
||||
docker_applied = list(docker_applied or [])
|
||||
|
||||
payload = {
|
||||
"egress": {
|
||||
"iface": egress_iface,
|
||||
@@ -121,6 +126,10 @@ def emit_json(
|
||||
"active": bool(wg_active),
|
||||
"applied": bool(wg_applied),
|
||||
},
|
||||
"docker": {
|
||||
"ifaces": docker_ifaces,
|
||||
"applied": docker_applied,
|
||||
},
|
||||
"dry_run": bool(dry_run),
|
||||
}
|
||||
|
||||
|
||||
170
src/automtu/persist.py
Normal file
170
src/automtu/persist.py
Normal 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)
|
||||
@@ -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)
|
||||
@@ -23,8 +23,7 @@ class TestCli(unittest.TestCase):
|
||||
"1472",
|
||||
"--pmtu-policy",
|
||||
"median",
|
||||
"--apply-egress-mtu",
|
||||
"--apply-wg-mtu",
|
||||
"--apply-all",
|
||||
"--wg-if",
|
||||
"wg0",
|
||||
"--wg-overhead",
|
||||
@@ -36,6 +35,8 @@ class TestCli(unittest.TestCase):
|
||||
"1372",
|
||||
"--force-egress-mtu",
|
||||
"1452",
|
||||
"--docker-if",
|
||||
"docker0,br-123",
|
||||
"--dry-run",
|
||||
"--print-mtu",
|
||||
"wg",
|
||||
@@ -49,14 +50,22 @@ class TestCli(unittest.TestCase):
|
||||
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.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_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.assertEqual(args.docker_if, ["docker0,br-123"])
|
||||
self.assertFalse(args.docker_no_user_bridges)
|
||||
|
||||
self.assertTrue(args.dry_run)
|
||||
self.assertEqual(args.print_mtu, "wg")
|
||||
self.assertFalse(args.print_json)
|
||||
|
||||
@@ -22,10 +22,18 @@ class TestCore(unittest.TestCase):
|
||||
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
|
||||
@@ -40,6 +48,7 @@ class TestCore(unittest.TestCase):
|
||||
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):
|
||||
@@ -52,10 +61,66 @@ class TestCore(unittest.TestCase):
|
||||
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_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,
|
||||
@@ -70,10 +135,18 @@ class TestCore(unittest.TestCase):
|
||||
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 (
|
||||
@@ -81,6 +154,7 @@ class TestCore(unittest.TestCase):
|
||||
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):
|
||||
|
||||
49
tests/unit/test_docker.py
Normal file
49
tests/unit/test_docker.py
Normal 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)
|
||||
@@ -1,4 +1,5 @@
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import automtu.net as net
|
||||
@@ -44,6 +45,20 @@ class TestNet(unittest.TestCase):
|
||||
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:
|
||||
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__":
|
||||
unittest.main(verbosity=2)
|
||||
|
||||
117
tests/unit/test_persist.py
Normal file
117
tests/unit/test_persist.py
Normal 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)
|
||||
Reference in New Issue
Block a user