8 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
a0db5f797f Release version 1.1.0
Some checks failed
CI / test-and-lint (push) Has been cancelled
2026-01-23 10:56:44 +01:00
fc5dee4dc0 feat: add Docker MTU apply support and docker-ordered persistence
https://chatgpt.com/share/69733e45-96ec-800f-9aad-0cac7306dedd
2026-01-23 10:53:29 +01:00
f534d025d0 feat(persist): add systemd persistence backend with uninstall
https://chatgpt.com/share/69733e45-96ec-800f-9aad-0cac7306dedd
2026-01-23 10:23:59 +01:00
77da7a4acb Added Mirrors and ignored build artefacts
Some checks failed
CI / test-and-lint (push) Has been cancelled
2026-01-21 19:18:07 +01:00
24 changed files with 882 additions and 59 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)"
]
}
}

3
.gitignore vendored
View File

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

View File

@@ -1,3 +1,19 @@
# 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
* * 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.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

@@ -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

@@ -2,9 +2,10 @@ 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 .net import ( from .net import (
default_route_uses_iface, default_route_uses_iface,
detect_egress_iface, detect_egress_iface,
@@ -27,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()])
@@ -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(
@@ -94,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)
@@ -115,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(
@@ -161,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:
@@ -192,6 +221,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 +271,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

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

@@ -0,0 +1,46 @@
from __future__ import annotations
import re
from .net import iface_exists, list_ifaces
_BRIDGE_RE = re.compile(r"^br-[0-9a-f]+$", re.IGNORECASE)
def _split_items(items: list[str] | None) -> 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: list[str] | None, *, 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

@@ -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,12 +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]:
"""
Return a sorted list of all network interfaces, netlink as fallback.
Plain files in /sys/class/net (e.g. bonding_masters) are not interfaces.
"""
try:
return sorted(
p.name for p in SYSFS_NET.iterdir() if p.is_symlink() or p.is_dir()
)
except OSError:
out = _run(["ip", "-o", "link", "show"])
return sorted(re.findall(r"^\d+:\s+([^:@\s]+)", out, flags=re.MULTILINE))
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:
@@ -38,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

@@ -1,21 +1,21 @@
# src/automtu/output.py
from __future__ import annotations 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
@@ -27,7 +27,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 +56,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)
@@ -67,22 +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: list[str] | None = None,
docker_applied: list[str] | None = None,
dry_run: bool, dry_run: bool,
) -> bool: ) -> bool:
""" """
@@ -91,6 +92,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 +125,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),
} }

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

@@ -0,0 +1,168 @@
from __future__ import annotations
import shlex
import shutil
import subprocess
from pathlib import Path
_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

@@ -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,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

@@ -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(
@@ -22,10 +56,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 +82,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 +95,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 +169,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 +188,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):
@@ -90,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)

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

@@ -0,0 +1,49 @@
import unittest
from unittest.mock import patch
from automtu import 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,7 +1,120 @@
import pathlib
import tempfile
import unittest import unittest
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):
@@ -44,6 +157,12 @@ 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:
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__": if __name__ == "__main__":
unittest.main(verbosity=2) unittest.main(verbosity=2)

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

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
from automtu import 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)

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