diff --git a/src/automtu/cli.py b/src/automtu/cli.py index 3565fcd..2e3171c 100644 --- a/src/automtu/cli.py +++ b/src/automtu/cli.py @@ -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", diff --git a/src/automtu/core.py b/src/automtu/core.py index ce3e456..9008758 100644 --- a/src/automtu/core.py +++ b/src/automtu/core.py @@ -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 diff --git a/src/automtu/docker.py b/src/automtu/docker.py new file mode 100644 index 0000000..3930361 --- /dev/null +++ b/src/automtu/docker.py @@ -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)) diff --git a/src/automtu/net.py b/src/automtu/net.py index b7aefe2..672a898 100644 --- a/src/automtu/net.py +++ b/src/automtu/net.py @@ -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()) diff --git a/src/automtu/output.py b/src/automtu/output.py index c4390a9..9a1c5e3 100644 --- a/src/automtu/output.py +++ b/src/automtu/output.py @@ -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), } diff --git a/src/automtu/persist.py b/src/automtu/persist.py index cf75c21..1dc631f 100644 --- a/src/automtu/persist.py +++ b/src/automtu/persist.py @@ -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) diff --git a/tests/unit/test_cli.py b/tests/unit/test_cli.py index 80be5f1..d0307e1 100644 --- a/tests/unit/test_cli.py +++ b/tests/unit/test_cli.py @@ -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) diff --git a/tests/unit/test_core.py b/tests/unit/test_core.py index af5e03b..5cb9915 100644 --- a/tests/unit/test_core.py +++ b/tests/unit/test_core.py @@ -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): diff --git a/tests/unit/test_docker.py b/tests/unit/test_docker.py new file mode 100644 index 0000000..b76665b --- /dev/null +++ b/tests/unit/test_docker.py @@ -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) diff --git a/tests/unit/test_net.py b/tests/unit/test_net.py index 8575094..e293f46 100644 --- a/tests/unit/test_net.py +++ b/tests/unit/test_net.py @@ -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) diff --git a/tests/unit/test_persist.py b/tests/unit/test_persist.py index ba4cb87..b3282ce 100644 --- a/tests/unit/test_persist.py +++ b/tests/unit/test_persist.py @@ -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__":