Compare commits
4 Commits
v1.0.0
...
a0db5f797f
| Author | SHA1 | Date | |
|---|---|---|---|
| a0db5f797f | |||
| fc5dee4dc0 | |||
| f534d025d0 | |||
| 77da7a4acb |
3
.gitignore
vendored
3
.gitignore
vendored
@@ -1,2 +1,5 @@
|
|||||||
*.pyc
|
*.pyc
|
||||||
__pycache__
|
__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
|
## [1.0.0] - 2026-01-21
|
||||||
|
|
||||||
* 🥳 Official Release
|
* 🥳 Official Release
|
||||||
|
|||||||
4
MIRRORS
Normal file
4
MIRRORS
Normal 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/
|
||||||
3
Makefile
3
Makefile
@@ -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
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "automtu"
|
name = "automtu"
|
||||||
version = "1.0.0"
|
version = "1.1.0"
|
||||||
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
|
description = "Auto-detect egress interface, probe Path MTU, and apply MTU (WireGuard/egress)."
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
|
|||||||
@@ -1,2 +0,0 @@
|
|||||||
__all__ = ["__version__"]
|
|
||||||
__version__ = "0.1.0"
|
|
||||||
|
|||||||
@@ -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",
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import sys
|
|||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from typing import Iterable, Optional
|
from typing import Iterable, Optional
|
||||||
|
|
||||||
|
from .docker import detect_docker_ifaces
|
||||||
from .net import (
|
from .net import (
|
||||||
default_route_uses_iface,
|
default_route_uses_iface,
|
||||||
detect_egress_iface,
|
detect_egress_iface,
|
||||||
@@ -46,6 +47,12 @@ def _choose(values: Iterable[int], policy: str) -> int:
|
|||||||
|
|
||||||
|
|
||||||
def run_automtu(args) -> int:
|
def run_automtu(args) -> int:
|
||||||
|
# Expand apply-all -> set apply flags
|
||||||
|
if getattr(args, "apply_all", False):
|
||||||
|
args.apply_egress_mtu = True
|
||||||
|
args.apply_wg_mtu = True
|
||||||
|
args.apply_docker_mtu = True
|
||||||
|
|
||||||
mode = OutputMode(
|
mode = OutputMode(
|
||||||
print_mtu=getattr(args, "print_mtu", None),
|
print_mtu=getattr(args, "print_mtu", None),
|
||||||
print_json=bool(getattr(args, "print_json", False)),
|
print_json=bool(getattr(args, "print_json", False)),
|
||||||
@@ -57,14 +64,32 @@ def run_automtu(args) -> int:
|
|||||||
|
|
||||||
log = Logger(mode.machine).log
|
log = Logger(mode.machine).log
|
||||||
|
|
||||||
# Root is only needed if we actually change something (without --dry-run).
|
|
||||||
needs_root = bool(
|
needs_root = bool(
|
||||||
args.apply_egress_mtu
|
getattr(args, "apply_egress_mtu", False)
|
||||||
or args.apply_wg_mtu
|
or getattr(args, "apply_wg_mtu", False)
|
||||||
or (args.force_egress_mtu is not None)
|
or getattr(args, "apply_docker_mtu", False)
|
||||||
|
or (getattr(args, "force_egress_mtu", None) is not None)
|
||||||
|
or (getattr(args, "persist", None) is not None)
|
||||||
)
|
)
|
||||||
require_root(dry=args.dry_run, needs_root=needs_root)
|
require_root(dry=args.dry_run, needs_root=needs_root)
|
||||||
|
|
||||||
|
# Persistence mode: install/uninstall persistence mechanism and exit.
|
||||||
|
if getattr(args, "persist", None):
|
||||||
|
if args.persist == "systemd":
|
||||||
|
from .persist import persist_systemd, uninstall_systemd
|
||||||
|
|
||||||
|
if getattr(args, "uninstall", False):
|
||||||
|
uninstall_systemd(dry=args.dry_run)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
persist_systemd(sys.argv, dry=args.dry_run)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
print(
|
||||||
|
f"[automtu][ERROR] Unknown persist backend: {args.persist}", file=sys.stderr
|
||||||
|
)
|
||||||
|
return 4
|
||||||
|
|
||||||
egress = args.egress_if or detect_egress_iface(ignore_vpn=not args.prefer_wg_egress)
|
egress = args.egress_if or detect_egress_iface(ignore_vpn=not args.prefer_wg_egress)
|
||||||
if not egress:
|
if not egress:
|
||||||
print(
|
print(
|
||||||
@@ -192,6 +217,29 @@ def run_automtu(args) -> int:
|
|||||||
else:
|
else:
|
||||||
log("[automtu] INFO: Not applying WireGuard MTU (use --apply-wg-mtu).")
|
log("[automtu] INFO: Not applying WireGuard MTU (use --apply-wg-mtu).")
|
||||||
|
|
||||||
|
# Apply Docker MTU (optional)
|
||||||
|
docker_ifaces = detect_docker_ifaces(
|
||||||
|
getattr(args, "docker_if", None),
|
||||||
|
include_user_bridges=not bool(getattr(args, "docker_no_user_bridges", False)),
|
||||||
|
)
|
||||||
|
docker_applied: list[str] = []
|
||||||
|
|
||||||
|
if args.apply_docker_mtu:
|
||||||
|
if not docker_ifaces:
|
||||||
|
log("[automtu] INFO: No Docker interfaces detected for MTU apply.")
|
||||||
|
else:
|
||||||
|
log(
|
||||||
|
f"[automtu] Applying effective MTU {effective_mtu} to Docker ifaces: {', '.join(docker_ifaces)}"
|
||||||
|
)
|
||||||
|
for d in docker_ifaces:
|
||||||
|
if iface_exists(d):
|
||||||
|
set_iface_mtu(d, effective_mtu, args.dry_run)
|
||||||
|
docker_applied.append(d)
|
||||||
|
else:
|
||||||
|
log(
|
||||||
|
"[automtu] INFO: Not applying Docker MTU (use --apply-docker-mtu or --apply-all)."
|
||||||
|
)
|
||||||
|
|
||||||
# Machine-readable outputs
|
# Machine-readable outputs
|
||||||
if emit_single_number(
|
if emit_single_number(
|
||||||
mode, base_mtu=base_mtu, effective_mtu=effective_mtu, wg_mtu=wg_mtu
|
mode, base_mtu=base_mtu, effective_mtu=effective_mtu, wg_mtu=wg_mtu
|
||||||
@@ -219,6 +267,8 @@ def run_automtu(args) -> int:
|
|||||||
wg_present=wg_present,
|
wg_present=wg_present,
|
||||||
wg_active=wg_active,
|
wg_active=wg_active,
|
||||||
wg_applied=wg_applied,
|
wg_applied=wg_applied,
|
||||||
|
docker_ifaces=docker_ifaces,
|
||||||
|
docker_applied=docker_applied,
|
||||||
dry_run=bool(args.dry_run),
|
dry_run=bool(args.dry_run),
|
||||||
):
|
):
|
||||||
return 0
|
return 0
|
||||||
|
|||||||
48
src/automtu/docker.py
Normal file
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()
|
return pathlib.Path(f"/sys/class/net/{iface}").exists()
|
||||||
|
|
||||||
|
|
||||||
|
def list_ifaces() -> list[str]:
|
||||||
|
"""
|
||||||
|
Return a sorted list of all network interfaces visible under /sys/class/net.
|
||||||
|
"""
|
||||||
|
base = pathlib.Path("/sys/class/net")
|
||||||
|
if not base.exists():
|
||||||
|
return []
|
||||||
|
names: list[str] = []
|
||||||
|
for p in base.iterdir():
|
||||||
|
if p.is_dir():
|
||||||
|
names.append(p.name)
|
||||||
|
names.sort()
|
||||||
|
return names
|
||||||
|
|
||||||
|
|
||||||
def read_iface_mtu(iface: str) -> int:
|
def read_iface_mtu(iface: str) -> int:
|
||||||
return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip())
|
return int(pathlib.Path(f"/sys/class/net/{iface}/mtu").read_text().strip())
|
||||||
|
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
# src/automtu/output.py
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import json
|
import json
|
||||||
@@ -27,7 +28,7 @@ class Logger:
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, machine_mode: bool) -> None:
|
def __init__(self, machine_mode: bool) -> None:
|
||||||
self._machine = machine_mode
|
self._machine = bool(machine_mode)
|
||||||
|
|
||||||
def log(self, msg: str) -> None:
|
def log(self, msg: str) -> None:
|
||||||
if self._machine:
|
if self._machine:
|
||||||
@@ -56,7 +57,6 @@ def emit_single_number(
|
|||||||
print(int(wg_mtu))
|
print(int(wg_mtu))
|
||||||
return True
|
return True
|
||||||
|
|
||||||
# Should never happen due to argparse choices
|
|
||||||
print("[automtu][ERROR] Invalid --print-mtu value.", file=sys.stderr)
|
print("[automtu][ERROR] Invalid --print-mtu value.", file=sys.stderr)
|
||||||
raise SystemExit(4)
|
raise SystemExit(4)
|
||||||
|
|
||||||
@@ -83,6 +83,8 @@ def emit_json(
|
|||||||
wg_present: bool,
|
wg_present: bool,
|
||||||
wg_active: bool,
|
wg_active: bool,
|
||||||
wg_applied: bool,
|
wg_applied: bool,
|
||||||
|
docker_ifaces: Optional[list[str]] = None,
|
||||||
|
docker_applied: Optional[list[str]] = None,
|
||||||
dry_run: bool,
|
dry_run: bool,
|
||||||
) -> bool:
|
) -> bool:
|
||||||
"""
|
"""
|
||||||
@@ -91,6 +93,9 @@ def emit_json(
|
|||||||
if not mode.print_json:
|
if not mode.print_json:
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
docker_ifaces = list(docker_ifaces or [])
|
||||||
|
docker_applied = list(docker_applied or [])
|
||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"egress": {
|
"egress": {
|
||||||
"iface": egress_iface,
|
"iface": egress_iface,
|
||||||
@@ -121,6 +126,10 @@ def emit_json(
|
|||||||
"active": bool(wg_active),
|
"active": bool(wg_active),
|
||||||
"applied": bool(wg_applied),
|
"applied": bool(wg_applied),
|
||||||
},
|
},
|
||||||
|
"docker": {
|
||||||
|
"ifaces": docker_ifaces,
|
||||||
|
"applied": docker_applied,
|
||||||
|
},
|
||||||
"dry_run": bool(dry_run),
|
"dry_run": bool(dry_run),
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
170
src/automtu/persist.py
Normal file
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",
|
"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)
|
||||||
|
|||||||
@@ -22,10 +22,18 @@ class TestCore(unittest.TestCase):
|
|||||||
pmtu_policy="min",
|
pmtu_policy="min",
|
||||||
apply_egress_mtu=True,
|
apply_egress_mtu=True,
|
||||||
apply_wg_mtu=True,
|
apply_wg_mtu=True,
|
||||||
|
apply_docker_mtu=False,
|
||||||
|
apply_all=False,
|
||||||
|
docker_if=None,
|
||||||
|
docker_no_user_bridges=False,
|
||||||
wg_if="wg0",
|
wg_if="wg0",
|
||||||
wg_overhead=80,
|
wg_overhead=80,
|
||||||
wg_min=1280,
|
wg_min=1280,
|
||||||
set_wg_mtu=None,
|
set_wg_mtu=None,
|
||||||
|
persist=None,
|
||||||
|
uninstall=False,
|
||||||
|
print_mtu=None,
|
||||||
|
print_json=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
# PMTU probes: 1452 and 1500 -> min policy => 1452, effective=min(base(1500),1452)=1452
|
# PMTU probes: 1452 and 1500 -> min policy => 1452, effective=min(base(1500),1452)=1452
|
||||||
@@ -40,6 +48,7 @@ class TestCore(unittest.TestCase):
|
|||||||
patch("automtu.core.wg_is_active", return_value=False),
|
patch("automtu.core.wg_is_active", return_value=False),
|
||||||
patch("automtu.core.wg_peer_endpoints", return_value=[]),
|
patch("automtu.core.wg_peer_endpoints", return_value=[]),
|
||||||
patch("automtu.core.default_route_uses_iface", return_value=False),
|
patch("automtu.core.default_route_uses_iface", return_value=False),
|
||||||
|
patch("automtu.core.detect_docker_ifaces", return_value=[]),
|
||||||
):
|
):
|
||||||
buf = io.StringIO()
|
buf = io.StringIO()
|
||||||
with redirect_stdout(buf):
|
with redirect_stdout(buf):
|
||||||
@@ -52,10 +61,66 @@ class TestCore(unittest.TestCase):
|
|||||||
self.assertIn("Selected Path MTU (policy=min): 1452", s)
|
self.assertIn("Selected Path MTU (policy=min): 1452", s)
|
||||||
self.assertIn("Computed wg0 MTU: 1372", s)
|
self.assertIn("Computed wg0 MTU: 1372", s)
|
||||||
|
|
||||||
# apply egress and wg are both true -> two calls in order
|
|
||||||
mock_set.assert_any_call("eth0", 1452, True)
|
mock_set.assert_any_call("eth0", 1452, True)
|
||||||
mock_set.assert_any_call("wg0", 1372, True)
|
mock_set.assert_any_call("wg0", 1372, True)
|
||||||
|
|
||||||
|
def test_run_automtu_apply_all_includes_docker_bridge(self) -> None:
|
||||||
|
args = SimpleNamespace(
|
||||||
|
dry_run=True,
|
||||||
|
egress_if="eth0",
|
||||||
|
prefer_wg_egress=False,
|
||||||
|
force_egress_mtu=None,
|
||||||
|
pmtu_target=None,
|
||||||
|
auto_pmtu_from_wg=False,
|
||||||
|
pmtu_min_payload=1200,
|
||||||
|
pmtu_max_payload=1472,
|
||||||
|
pmtu_timeout=1.0,
|
||||||
|
pmtu_policy="min",
|
||||||
|
apply_egress_mtu=False,
|
||||||
|
apply_wg_mtu=False,
|
||||||
|
apply_docker_mtu=False,
|
||||||
|
apply_all=True, # expands in core()
|
||||||
|
docker_if=None,
|
||||||
|
docker_no_user_bridges=False,
|
||||||
|
wg_if="wg0",
|
||||||
|
wg_overhead=80,
|
||||||
|
wg_min=1280,
|
||||||
|
set_wg_mtu=None,
|
||||||
|
persist=None,
|
||||||
|
uninstall=False,
|
||||||
|
print_mtu=None,
|
||||||
|
print_json=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("automtu.core.require_root", return_value=None),
|
||||||
|
patch(
|
||||||
|
"automtu.core.iface_exists",
|
||||||
|
side_effect=lambda name: name in {"eth0", "wg0", "docker0", "br-abc"},
|
||||||
|
),
|
||||||
|
patch("automtu.core.read_iface_mtu", return_value=1500),
|
||||||
|
patch("automtu.core.set_iface_mtu") as mock_set,
|
||||||
|
patch("automtu.core.wg_is_active", return_value=True),
|
||||||
|
patch(
|
||||||
|
"automtu.core.detect_docker_ifaces", return_value=["docker0", "br-abc"]
|
||||||
|
),
|
||||||
|
):
|
||||||
|
buf = io.StringIO()
|
||||||
|
with redirect_stdout(buf):
|
||||||
|
rc = run_automtu(args)
|
||||||
|
|
||||||
|
self.assertEqual(rc, 0)
|
||||||
|
|
||||||
|
# egress applied
|
||||||
|
mock_set.assert_any_call("eth0", 1500, True)
|
||||||
|
|
||||||
|
# wg applied (1500-80=1420)
|
||||||
|
mock_set.assert_any_call("wg0", 1420, True)
|
||||||
|
|
||||||
|
# docker applied
|
||||||
|
mock_set.assert_any_call("docker0", 1500, True)
|
||||||
|
mock_set.assert_any_call("br-abc", 1500, True)
|
||||||
|
|
||||||
def test_run_automtu_does_not_apply_wg_without_flag(self) -> None:
|
def test_run_automtu_does_not_apply_wg_without_flag(self) -> None:
|
||||||
args = SimpleNamespace(
|
args = SimpleNamespace(
|
||||||
dry_run=True,
|
dry_run=True,
|
||||||
@@ -70,10 +135,18 @@ class TestCore(unittest.TestCase):
|
|||||||
pmtu_policy="min",
|
pmtu_policy="min",
|
||||||
apply_egress_mtu=False,
|
apply_egress_mtu=False,
|
||||||
apply_wg_mtu=False,
|
apply_wg_mtu=False,
|
||||||
|
apply_docker_mtu=False,
|
||||||
|
apply_all=False,
|
||||||
|
docker_if=None,
|
||||||
|
docker_no_user_bridges=False,
|
||||||
wg_if="wg0",
|
wg_if="wg0",
|
||||||
wg_overhead=80,
|
wg_overhead=80,
|
||||||
wg_min=1280,
|
wg_min=1280,
|
||||||
set_wg_mtu=None,
|
set_wg_mtu=None,
|
||||||
|
persist=None,
|
||||||
|
uninstall=False,
|
||||||
|
print_mtu=None,
|
||||||
|
print_json=False,
|
||||||
)
|
)
|
||||||
|
|
||||||
with (
|
with (
|
||||||
@@ -81,6 +154,7 @@ class TestCore(unittest.TestCase):
|
|||||||
patch("automtu.core.iface_exists", return_value=True),
|
patch("automtu.core.iface_exists", return_value=True),
|
||||||
patch("automtu.core.read_iface_mtu", return_value=1500),
|
patch("automtu.core.read_iface_mtu", return_value=1500),
|
||||||
patch("automtu.core.set_iface_mtu") as mock_set,
|
patch("automtu.core.set_iface_mtu") as mock_set,
|
||||||
|
patch("automtu.core.detect_docker_ifaces", return_value=[]),
|
||||||
):
|
):
|
||||||
buf = io.StringIO()
|
buf = io.StringIO()
|
||||||
with redirect_stdout(buf):
|
with redirect_stdout(buf):
|
||||||
|
|||||||
49
tests/unit/test_docker.py
Normal file
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
|
import unittest
|
||||||
|
from pathlib import Path
|
||||||
from unittest.mock import patch
|
from unittest.mock import patch
|
||||||
|
|
||||||
import automtu.net as net
|
import automtu.net as net
|
||||||
@@ -44,6 +45,20 @@ class TestNet(unittest.TestCase):
|
|||||||
self.assertTrue(net.default_route_uses_iface("eth0"))
|
self.assertTrue(net.default_route_uses_iface("eth0"))
|
||||||
self.assertFalse(net.default_route_uses_iface("wg0"))
|
self.assertFalse(net.default_route_uses_iface("wg0"))
|
||||||
|
|
||||||
|
def test_list_ifaces_returns_sorted_names(self) -> None:
|
||||||
|
fake = [
|
||||||
|
Path("/sys/class/net/eth0"),
|
||||||
|
Path("/sys/class/net/lo"),
|
||||||
|
Path("/sys/class/net/wg0"),
|
||||||
|
]
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("automtu.net.pathlib.Path.exists", return_value=True),
|
||||||
|
patch("automtu.net.pathlib.Path.iterdir", return_value=fake),
|
||||||
|
patch.object(Path, "is_dir", return_value=True),
|
||||||
|
):
|
||||||
|
self.assertEqual(net.list_ifaces(), ["eth0", "lo", "wg0"])
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
unittest.main(verbosity=2)
|
unittest.main(verbosity=2)
|
||||||
|
|||||||
117
tests/unit/test_persist.py
Normal file
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