Merge 177ec93555 into 8c99ac78b7

src/netsleuth/models.py

@@ -287,6 +287,17 @@ class TraceConfig:
     max_hops: int = 30
 
 
+@dataclass(frozen=True)
+class TraceCommandOpts:
+    backend: str
+    binary: str
+    target: str
+    max_hops: int
+    cycles: int
+    os_name: str
+    source_ip: str | None = None
+
+
 @dataclass
 class TraceHop:
     ttl: int

src/netsleuth/probes/traceroute.py

@@ -9,7 +9,7 @@ import threading
 from collections.abc import Awaitable, Callable
 from typing import TypeVar
 
-from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult
+from netsleuth.models import Capabilities, TraceCommandOpts, TraceConfig, TraceHop, TraceResult
 from netsleuth.probes.icmp_win import IcmpReply, classify_status
 from netsleuth.traceparse import build_trace_result, finalize_hop
 
@@ -63,30 +63,22 @@ def filter_trace_tiers(order: list[str], os_name: str, forced_v4: bool) -> tuple
     return [t for t in order if t != "system_traceroute"], ["system_traceroute"]
 
 
-def trace_argv(
+def trace_argv(opts: TraceCommandOpts) -> list[str]:
-    backend: str,
+    if opts.backend == "mtr_json":
-    binary: str,
+        args = [opts.binary, "--json", "--report-cycles", str(opts.cycles), "--max-ttl", str(opts.max_hops)]
-    target: str,
+        if opts.source_ip:
-    max_hops: int,
+            args += ["--address", opts.source_ip]
-    cycles: int,
+        args.append(opts.target)
-    os_name: str,
-    source_ip: str | None = None,
-) -> list[str]:
-    if backend == "mtr_json":
-        args = [binary, "--json", "--report-cycles", str(cycles), "--max-ttl", str(max_hops)]
-        if source_ip:
-            args += ["--address", source_ip]
-        args.append(target)
         return args
-    if backend == "system_traceroute":
+    if opts.backend == "system_traceroute":
-        if os_name == "Windows":
+        if opts.os_name == "Windows":
-            return [binary, "-h", str(max_hops), "-w", "2", target]
+            return [opts.binary, "-h", str(opts.max_hops), "-w", "2", opts.target]
-        args = [binary, "-m", str(max_hops), "-w", "2"]
+        args = [opts.binary, "-m", str(opts.max_hops), "-w", "2"]
-        if source_ip:
+        if opts.source_ip:
-            args += ["-s", source_ip]
+            args += ["-s", opts.source_ip]
-        args.append(target)
+        args.append(opts.target)
         return args
-    raise ValueError(f"unsupported backend for trace_argv: {backend!r}")
+    raise ValueError(f"unsupported backend for trace_argv: {opts.backend!r}")
 
 
 async def run_cascade(tiers: list[tuple[str, Callable[[], Awaitable[TraceResult]]]]) -> TraceResult:
@@ -139,9 +131,16 @@ def _console_encoding() -> str:
 async def _tier_mtr(
     target: str, binary: str, cycles: int, max_hops: int, timeout: float, source_ip: str | None = None
 ) -> TraceResult:
-    text = await _run(
+    opts = TraceCommandOpts(
-        trace_argv("mtr_json", binary, target, max_hops, cycles, platform.system(), source_ip), timeout
+        backend="mtr_json",
+        binary=binary,
+        target=target,
+        max_hops=max_hops,
+        cycles=cycles,
+        os_name=platform.system(),
+        source_ip=source_ip,
     )
+    text = await _run(trace_argv(opts), timeout)
     payload = json.loads(text)
     report = payload.get("report") or {}
     hops = []
@@ -212,7 +211,16 @@ async def _tier_system(
     target: str, binary: str, max_hops: int, timeout: float, source_ip: str | None = None
 ) -> TraceResult:
     os_name = platform.system()
-    args = trace_argv("system_traceroute", binary, target, max_hops, 1, os_name, source_ip)
+    opts = TraceCommandOpts(
+        backend="system_traceroute",
+        binary=binary,
+        target=target,
+        max_hops=max_hops,
+        cycles=1,
+        os_name=os_name,
+        source_ip=source_ip,
+    )
+    args = trace_argv(opts)
     text = await _run(args, timeout)
     config = TraceConfig(target=target, resolved_ip=None, max_hops=max_hops)
     return build_trace_result(text, os_name, config)

tests/test_cli.py

@@ -1,5 +1,4 @@
-import os
+from pathlib import PurePosixPath
-from pathlib import Path, PurePosixPath
 from unittest.mock import MagicMock, patch
 
 from netsleuth.cli import _dedupe, _os_timezone, parse_target

tests/test_netinfo.py

@@ -111,7 +111,6 @@ def test_degradation_note_on_windows_mentions_the_api_rather_than_sysctl():
 
 import pytest
 
-from unittest.mock import patch
 
 from netsleuth.netinfo import iface_for_ip, is_tunnel_iface, mtu_anomaly, primary_interface_ip
 
@@ -244,7 +243,6 @@ class MockStats:
 
 from unittest.mock import patch
 from netsleuth.netinfo import collect_local_net
-from netsleuth.models import LocalNet
 
 @patch("netsleuth.netinfo.primary_interface_ip")
 @patch("netsleuth.netinfo.psutil.net_if_addrs")

tests/test_traceroute_cascade.py

@@ -5,7 +5,7 @@ import threading
 
 import pytest
 
-from netsleuth.models import Capabilities, TraceHop, TraceResult
+from netsleuth.models import Capabilities, TraceCommandOpts, TraceHop, TraceResult
 from netsleuth.probes.icmp_win import (
     IP_REQ_TIMED_OUT,
     IP_SUCCESS,
@@ -245,7 +245,8 @@ def test_filter_trace_tiers_is_a_noop_when_system_traceroute_is_absent():
 
 
 def test_trace_argv_mtr_adds_address_flag_when_source_given():
-    args = trace_argv("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux", source_ip="192.168.1.34")
+    opts = TraceCommandOpts("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux", source_ip="192.168.1.34")
+    args = trace_argv(opts)
     assert args == [
         "/usr/bin/mtr", "--json", "--report-cycles", "10", "--max-ttl", "30",
         "--address", "192.168.1.34", "1.1.1.1",
@@ -253,22 +254,25 @@ def test_trace_argv_mtr_adds_address_flag_when_source_given():
 
 
 def test_trace_argv_mtr_omits_address_flag_without_a_source():
-    args = trace_argv("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux")
+    opts = TraceCommandOpts("mtr_json", "/usr/bin/mtr", "1.1.1.1", max_hops=30, cycles=10, os_name="Linux")
+    args = trace_argv(opts)
     assert "--address" not in args
 
 
 def test_trace_argv_unix_traceroute_adds_source_flag():
-    args = trace_argv(
+    opts = TraceCommandOpts(
         "system_traceroute", "/usr/bin/traceroute", "1.1.1.1", max_hops=30, cycles=1,
         os_name="Linux", source_ip="192.168.1.34",
     )
+    args = trace_argv(opts)
     assert args == ["/usr/bin/traceroute", "-m", "30", "-w", "2", "-s", "192.168.1.34", "1.1.1.1"]
 
 
 def test_trace_argv_windows_tracert_never_gets_a_source_flag():
-    args = trace_argv(
+    opts = TraceCommandOpts(
         "system_traceroute", "tracert", "1.1.1.1", max_hops=30, cycles=1,
         os_name="Windows", source_ip="192.168.1.34",
     )
+    args = trace_argv(opts)
     assert args == ["tracert", "-h", "30", "-w", "2", "1.1.1.1"]
     assert "-s" not in args and "-S" not in args