diff --git a/src/netsleuth/models.py b/src/netsleuth/models.py index 0e502ae..565abfb 100644 --- a/src/netsleuth/models.py +++ b/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 diff --git a/src/netsleuth/probes/traceroute.py b/src/netsleuth/probes/traceroute.py index b490202..9821359 100644 --- a/src/netsleuth/probes/traceroute.py +++ b/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( - backend: str, - binary: str, - target: str, - max_hops: int, - cycles: int, - 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) +def trace_argv(opts: TraceCommandOpts) -> list[str]: + if opts.backend == "mtr_json": + args = [opts.binary, "--json", "--report-cycles", str(opts.cycles), "--max-ttl", str(opts.max_hops)] + if opts.source_ip: + args += ["--address", opts.source_ip] + args.append(opts.target) return args - if backend == "system_traceroute": - if os_name == "Windows": - return [binary, "-h", str(max_hops), "-w", "2", target] - args = [binary, "-m", str(max_hops), "-w", "2"] - if source_ip: - args += ["-s", source_ip] - args.append(target) + if opts.backend == "system_traceroute": + if opts.os_name == "Windows": + return [opts.binary, "-h", str(opts.max_hops), "-w", "2", opts.target] + args = [opts.binary, "-m", str(opts.max_hops), "-w", "2"] + if opts.source_ip: + args += ["-s", opts.source_ip] + 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( - trace_argv("mtr_json", binary, target, max_hops, cycles, platform.system(), source_ip), timeout + opts = TraceCommandOpts( + 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) diff --git a/tests/test_cli.py b/tests/test_cli.py index 0e72067..6b601b8 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,5 +1,4 @@ -import os -from pathlib import Path, PurePosixPath +from pathlib import PurePosixPath from unittest.mock import MagicMock, patch from netsleuth.cli import _dedupe, _os_timezone, parse_target diff --git a/tests/test_netinfo.py b/tests/test_netinfo.py index 7134cd9..7aa16da 100644 --- a/tests/test_netinfo.py +++ b/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") diff --git a/tests/test_traceroute_cascade.py b/tests/test_traceroute_cascade.py index f6fd693..f7111f9 100644 --- a/tests/test_traceroute_cascade.py +++ b/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