diff --git a/src/netsleuth/models.py b/src/netsleuth/models.py index 7430741..5a2111f 100644 --- a/src/netsleuth/models.py +++ b/src/netsleuth/models.py @@ -268,6 +268,14 @@ class PingResult: samples: list[float | None] = field(default_factory=list) +@dataclass +class TraceConfig: + target: str + resolved_ip: str | None + backend: str = "system_traceroute" + max_hops: int = 30 + + @dataclass class TraceHop: ttl: int diff --git a/src/netsleuth/probes/traceroute.py b/src/netsleuth/probes/traceroute.py index 437c669..4c19c67 100644 --- a/src/netsleuth/probes/traceroute.py +++ b/src/netsleuth/probes/traceroute.py @@ -8,7 +8,7 @@ import shutil import threading from typing import Awaitable, Callable, TypeVar -from netsleuth.models import Capabilities, TraceHop, TraceResult +from netsleuth.models import Capabilities, TraceConfig, TraceHop, TraceResult from netsleuth.probes.icmp_win import IcmpReply, classify_status from netsleuth.traceparse import build_trace_result, finalize_hop @@ -213,7 +213,8 @@ async def _tier_system( os_name = platform.system() args = trace_argv("system_traceroute", binary, target, max_hops, 1, os_name, source_ip) text = await _run(args, timeout) - return build_trace_result(text, os_name, target=target, resolved_ip=None, max_hops=max_hops) + config = TraceConfig(target=target, resolved_ip=None, max_hops=max_hops) + return build_trace_result(text, os_name, config) async def _tier_tcp_trace( diff --git a/src/netsleuth/traceparse.py b/src/netsleuth/traceparse.py index 5065fff..6de15e8 100644 --- a/src/netsleuth/traceparse.py +++ b/src/netsleuth/traceparse.py @@ -2,7 +2,7 @@ from __future__ import annotations import re -from netsleuth.models import TraceHop, TraceResult +from netsleuth.models import TraceConfig, TraceHop, TraceResult from netsleuth.stats import rtt_stats IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b") @@ -160,18 +160,15 @@ def parse_traceroute(text: str, os_name: str) -> list[TraceHop]: def build_trace_result( text: str, os_name: str, - target: str, - resolved_ip: str | None, - backend: str = "system_traceroute", - max_hops: int = 30, + config: TraceConfig, ) -> TraceResult: hops = parse_traceroute(text, os_name) - completed = bool(hops and resolved_ip and hops[-1].ip == resolved_ip) - max_hops_reached = bool(hops) and not completed and hops[-1].ttl >= max_hops + completed = bool(hops and config.resolved_ip and hops[-1].ip == config.resolved_ip) + max_hops_reached = bool(hops) and not completed and hops[-1].ttl >= config.max_hops return TraceResult( - target=target, - resolved_ip=resolved_ip, - backend=backend, + target=config.target, + resolved_ip=config.resolved_ip, + backend=config.backend, hops=hops, cycles=1, completed=completed, diff --git a/tests/test_traceparse.py b/tests/test_traceparse.py index 283f0fa..d085edd 100644 --- a/tests/test_traceparse.py +++ b/tests/test_traceparse.py @@ -185,6 +185,7 @@ def test_darwin_parser_tolerates_empty_input(): assert parse_darwin("") == [] +from netsleuth.models import TraceConfig from netsleuth.traceparse import build_trace_result, parse_traceroute @@ -206,9 +207,11 @@ def test_build_trace_result_marks_completion_when_the_target_is_reached(trace_fi result = build_trace_result( trace_fixture("linux", "gnu_basic.txt"), "Linux", - target="1.1.1.1", - resolved_ip="1.1.1.1", - max_hops=30, + TraceConfig( + target="1.1.1.1", + resolved_ip="1.1.1.1", + max_hops=30, + ), ) assert result.completed is True assert result.max_hops_reached is False @@ -221,9 +224,11 @@ def test_build_trace_result_flags_a_run_that_died_at_max_hops(trace_fixture): result = build_trace_result( trace_fixture("linux", "gnu_maxhops.txt"), "Linux", - target="203.0.113.200", - resolved_ip="203.0.113.200", - max_hops=5, + TraceConfig( + target="203.0.113.200", + resolved_ip="203.0.113.200", + max_hops=5, + ), ) assert result.completed is False assert result.max_hops_reached is True @@ -233,16 +238,18 @@ def test_build_trace_result_is_not_complete_when_the_last_hop_is_a_different_hos result = build_trace_result( trace_fixture("linux", "gnu_unreachable.txt"), "Linux", - target="203.0.113.9", - resolved_ip="203.0.113.9", - max_hops=30, + TraceConfig( + target="203.0.113.9", + resolved_ip="203.0.113.9", + max_hops=30, + ), ) assert result.completed is False assert result.max_hops_reached is False def test_build_trace_result_on_empty_output_is_a_well_formed_empty_result(): - result = build_trace_result("", "Linux", target="1.1.1.1", resolved_ip=None) + result = build_trace_result("", "Linux", TraceConfig(target="1.1.1.1", resolved_ip=None)) assert result.hops == [] assert result.completed is False assert result.max_hops_reached is False