kevinveenbirkenbach/automtu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Docker MTU apply support and docker-ordered persistence

Browse Source 
https://chatgpt.com/share/69733e45-96ec-800f-9aad-0cac7306dedd
This commit is contained in:
Kevin Veen-Birkenbach
2026-01-23 10:53:29 +01:00
parent f534d025d0
commit fc5dee4dc0
11 changed files with 432 additions and 101 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
src/automtu/cli.py
View File

@@ -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,
@@ -94,8 +119,8 @@ def build_parser() -> argparse.ArgumentParser:
# --- Persistence ---
ap.add_argument(
"--persist",
choices=["systemd"],
help="Persist MTU configuration across reboots (currently supported: systemd).",
choices=["systemd", "docker"],
help="Persist MTU configuration across reboots (supported: systemd, docker).",
)
ap.add_argument(
"--uninstall",

33
src/automtu/core.py
View File

@@ -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)),
@@ -60,6 +67,7 @@ def run_automtu(args) -> int:
needs_root = bool(
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)
)
@@ -209,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
@@ -236,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
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))

15
src/automtu/net.py
View File

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

13
src/automtu/output.py
View File

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

143
src/automtu/persist.py
View File

@@ -8,13 +8,14 @@ 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
- --persist=systemd
- --persist systemd|docker
- --persist=systemd|docker
- --uninstall
Keeps all other args as-is.
"""
@@ -50,10 +51,74 @@ def _resolve_exec(argv0: str) -> str:
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.
This provides a distro-agnostic persistence mechanism.
Adds docker ordering automatically if docker MTU is applied.
"""
if not argv:
raise ValueError("argv must not be empty")
@@ -64,56 +129,42 @@ def persist_systemd(argv: List[str], *, dry: bool) -> None:
exe = _resolve_exec(filtered[0])
args = [exe, *filtered[1:]]
execstart = shlex.join(args)
unit = f"""\
[Unit]
Description=Auto MTU via automtu
After=network-online.target
Wants=network-online.target
[Service]
Type=oneshot
ExecStart={execstart}
[Install]
WantedBy=multi-user.target
"""
if dry:
print(f"[automtu] DRY-RUN: would write systemd unit to {_SYSTEMD_UNIT_PATH}")
print(unit.rstrip())
print("[automtu] DRY-RUN: would run: systemctl daemon-reload")
print("[automtu] DRY-RUN: would run: systemctl enable automtu.service")
return
_SYSTEMD_UNIT_PATH.write_text(unit)
subprocess.run(["systemctl", "daemon-reload"], check=True)
subprocess.run(["systemctl", "enable", "automtu.service"], check=True)
print("[automtu] Installed and enabled systemd service: automtu.service")
print("[automtu] Tip: run 'systemctl start automtu.service' to apply immediately.")
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 systemd persistence backend:
- disable automtu.service
- remove unit file (if present)
- daemon-reload
Uninstall the base systemd persistence backend.
"""
if dry:
print("[automtu] DRY-RUN: would run: systemctl disable automtu.service")
print(f"[automtu] DRY-RUN: would remove: {_SYSTEMD_UNIT_PATH} (if exists)")
print("[automtu] DRY-RUN: would run: systemctl daemon-reload")
return
_uninstall_unit(_SYSTEMD_UNIT_PATH, dry=dry)
subprocess.run(["systemctl", "disable", "automtu.service"], check=True)
if _SYSTEMD_UNIT_PATH.exists():
_SYSTEMD_UNIT_PATH.unlink()
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")
subprocess.run(["systemctl", "daemon-reload"], check=True)
print("[automtu] Uninstalled systemd service: automtu.service")
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)

17
tests/unit/test_cli.py
View File

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

76
tests/unit/test_core.py
View File

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

15
tests/unit/test_net.py
View File

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

93
tests/unit/test_persist.py
View File

@@ -23,7 +23,7 @@ class TestPersist(unittest.TestCase):
)
def test_strip_persist_args_removes_uninstall(self) -> None:
argv = ["automtu", "--persist", "systemd", "--uninstall", "--apply-wg-mtu"]
argv = ["automtu", "--persist", "docker", "--uninstall", "--apply-wg-mtu"]
self.assertEqual(
persist._strip_persist_args(argv),
["automtu", "--apply-wg-mtu"],
@@ -52,6 +52,43 @@ class TestPersist(unittest.TestCase):
"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()
@@ -62,52 +99,18 @@ class TestPersist(unittest.TestCase):
self.assertIn("DRY-RUN", s)
self.assertIn("systemctl disable", s)
def test_uninstall_systemd_runs_disable_and_removes_unit_when_present(self) -> None:
fake_path = Path("/tmp/automtu.service")
calls = []
def fake_run(cmd, check=False): # type: ignore[no-untyped-def]
calls.append((tuple(cmd), bool(check)))
with (
patch("automtu.persist._SYSTEMD_UNIT_PATH", fake_path),
patch("automtu.persist.subprocess.run", side_effect=fake_run),
patch.object(Path, "exists", return_value=True),
patch.object(Path, "unlink", return_value=None),
def test_uninstall_docker_dry_run_prints_actions(self) -> None:
with patch(
"automtu.persist._DOCKER_SYSTEMD_UNIT_PATH",
Path("/tmp/automtu-docker.service"),
):
persist.uninstall_systemd(dry=False)
out = io.StringIO()
with redirect_stdout(out):
persist.uninstall_docker(dry=True)
# Must disable unit and reload daemon
self.assertIn((("systemctl", "disable", "automtu.service"), True), calls)
self.assertIn((("systemctl", "daemon-reload"), True), calls)
def test_persist_systemd_writes_unit_and_enables(self) -> None:
argv = ["automtu", "--apply-wg-mtu", "--persist", "systemd"]
writes = {"text": None}
calls = []
def fake_write_text(self, txt, *args, **kwargs): # type: ignore[no-untyped-def]
writes["text"] = txt
return len(txt)
def fake_run(cmd, check=False): # type: ignore[no-untyped-def]
calls.append((tuple(cmd), bool(check)))
with (
patch("automtu.persist.shutil.which", return_value="/usr/bin/automtu"),
patch("automtu.persist._SYSTEMD_UNIT_PATH", Path("/tmp/automtu.service")),
patch.object(Path, "write_text", new=fake_write_text),
patch("automtu.persist.subprocess.run", side_effect=fake_run),
):
persist.persist_systemd(argv, dry=False)
self.assertIsNotNone(writes["text"])
self.assertIn("ExecStart=/usr/bin/automtu --apply-wg-mtu", writes["text"] or "")
self.assertIn((("systemctl", "daemon-reload"), True), calls)
self.assertIn((("systemctl", "enable", "automtu.service"), True), calls)
s = out.getvalue()
self.assertIn("DRY-RUN", s)
self.assertIn("automtu-docker.service", s)
if __name__ == "__main__":