Kevin Veen-Birkenbach
6392131577
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>
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
|
|
from automtu.output import Logger, OutputMode, emit_json, emit_single_number
|
|
|
|
|
|
class TestOutput(unittest.TestCase):
|
|
def test_output_mode_validate_mutual_exclusive(self) -> None:
|
|
mode = OutputMode(print_mtu="effective", print_json=True)
|
|
self.assertIsNotNone(mode.validate())
|
|
|
|
mode_ok = OutputMode(print_mtu=None, print_json=True)
|
|
self.assertIsNone(mode_ok.validate())
|
|
|
|
def test_logger_routes_to_stderr_in_machine_mode(self) -> None:
|
|
log = Logger(machine_mode=True).log
|
|
|
|
out = io.StringIO()
|
|
err = io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
log("[automtu] hello")
|
|
|
|
self.assertEqual(out.getvalue(), "")
|
|
self.assertIn("hello", err.getvalue())
|
|
|
|
def test_emit_single_number_effective(self) -> None:
|
|
mode = OutputMode(print_mtu="effective", print_json=False)
|
|
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
emitted = emit_single_number(
|
|
mode, base_mtu=1500, effective_mtu=1452, wg_mtu=1372
|
|
)
|
|
|
|
self.assertTrue(emitted)
|
|
self.assertEqual(out.getvalue().strip(), "1452")
|
|
|
|
def test_emit_json_is_valid_and_contains_expected_fields(self) -> None:
|
|
mode = OutputMode(print_mtu=None, print_json=True)
|
|
|
|
out = io.StringIO()
|
|
with redirect_stdout(out):
|
|
emitted = emit_json(
|
|
mode,
|
|
egress_iface="eth0",
|
|
base_mtu=1500,
|
|
effective_mtu=1452,
|
|
egress_forced_mtu=None,
|
|
egress_applied=False,
|
|
pmtu_targets=["1.1.1.1", "8.8.8.8"],
|
|
pmtu_auto_targets_added=[],
|
|
pmtu_policy="min",
|
|
pmtu_chosen=1452,
|
|
pmtu_results={"1.1.1.1": 1452, "8.8.8.8": None},
|
|
wg_iface="wg0",
|
|
wg_mtu=1372,
|
|
wg_overhead=80,
|
|
wg_min=1280,
|
|
wg_set_mtu=None,
|
|
wg_clamped=False,
|
|
wg_present=True,
|
|
wg_active=False,
|
|
wg_applied=False,
|
|
dry_run=True,
|
|
)
|
|
|
|
self.assertTrue(emitted)
|
|
|
|
payload = json.loads(out.getvalue())
|
|
self.assertEqual(payload["egress"]["iface"], "eth0")
|
|
self.assertEqual(payload["egress"]["base_mtu"], 1500)
|
|
self.assertEqual(payload["egress"]["effective_mtu"], 1452)
|
|
|
|
self.assertEqual(payload["pmtu"]["policy"], "min")
|
|
self.assertEqual(payload["pmtu"]["chosen"], 1452)
|
|
self.assertEqual(payload["pmtu"]["results"]["1.1.1.1"], 1452)
|
|
self.assertIsNone(payload["pmtu"]["results"]["8.8.8.8"])
|
|
|
|
self.assertEqual(payload["wg"]["iface"], "wg0")
|
|
self.assertEqual(payload["wg"]["mtu"], 1372)
|
|
self.assertTrue(payload["dry_run"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|