group build_trace_result arguments into TraceConfig dataclass

This commit is contained in:
Andrew Pimenov 2026-08-09 16:08:52 +03:00
parent 4c290ec309
commit f59f860327
4 changed files with 35 additions and 22 deletions

View file

@ -268,6 +268,14 @@ class PingResult:
samples: list[float | None] = field(default_factory=list) 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 @dataclass
class TraceHop: class TraceHop:
ttl: int ttl: int

View file

@ -8,7 +8,7 @@ import shutil
import threading import threading
from typing import Awaitable, Callable, TypeVar 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.probes.icmp_win import IcmpReply, classify_status
from netsleuth.traceparse import build_trace_result, finalize_hop from netsleuth.traceparse import build_trace_result, finalize_hop
@ -213,7 +213,8 @@ async def _tier_system(
os_name = platform.system() os_name = platform.system()
args = trace_argv("system_traceroute", binary, target, max_hops, 1, os_name, source_ip) args = trace_argv("system_traceroute", binary, target, max_hops, 1, os_name, source_ip)
text = await _run(args, timeout) 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( async def _tier_tcp_trace(

View file

@ -2,7 +2,7 @@ from __future__ import annotations
import re import re
from netsleuth.models import TraceHop, TraceResult from netsleuth.models import TraceConfig, TraceHop, TraceResult
from netsleuth.stats import rtt_stats from netsleuth.stats import rtt_stats
IPV4_RE = re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b") 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( def build_trace_result(
text: str, text: str,
os_name: str, os_name: str,
target: str, config: TraceConfig,
resolved_ip: str | None,
backend: str = "system_traceroute",
max_hops: int = 30,
) -> TraceResult: ) -> TraceResult:
hops = parse_traceroute(text, os_name) hops = parse_traceroute(text, os_name)
completed = bool(hops and resolved_ip and hops[-1].ip == resolved_ip) 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 >= max_hops max_hops_reached = bool(hops) and not completed and hops[-1].ttl >= config.max_hops
return TraceResult( return TraceResult(
target=target, target=config.target,
resolved_ip=resolved_ip, resolved_ip=config.resolved_ip,
backend=backend, backend=config.backend,
hops=hops, hops=hops,
cycles=1, cycles=1,
completed=completed, completed=completed,

View file

@ -185,6 +185,7 @@ def test_darwin_parser_tolerates_empty_input():
assert parse_darwin("") == [] assert parse_darwin("") == []
from netsleuth.models import TraceConfig
from netsleuth.traceparse import build_trace_result, parse_traceroute 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( result = build_trace_result(
trace_fixture("linux", "gnu_basic.txt"), trace_fixture("linux", "gnu_basic.txt"),
"Linux", "Linux",
TraceConfig(
target="1.1.1.1", target="1.1.1.1",
resolved_ip="1.1.1.1", resolved_ip="1.1.1.1",
max_hops=30, max_hops=30,
),
) )
assert result.completed is True assert result.completed is True
assert result.max_hops_reached is False 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( result = build_trace_result(
trace_fixture("linux", "gnu_maxhops.txt"), trace_fixture("linux", "gnu_maxhops.txt"),
"Linux", "Linux",
TraceConfig(
target="203.0.113.200", target="203.0.113.200",
resolved_ip="203.0.113.200", resolved_ip="203.0.113.200",
max_hops=5, max_hops=5,
),
) )
assert result.completed is False assert result.completed is False
assert result.max_hops_reached is True 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( result = build_trace_result(
trace_fixture("linux", "gnu_unreachable.txt"), trace_fixture("linux", "gnu_unreachable.txt"),
"Linux", "Linux",
TraceConfig(
target="203.0.113.9", target="203.0.113.9",
resolved_ip="203.0.113.9", resolved_ip="203.0.113.9",
max_hops=30, max_hops=30,
),
) )
assert result.completed is False assert result.completed is False
assert result.max_hops_reached is False assert result.max_hops_reached is False
def test_build_trace_result_on_empty_output_is_a_well_formed_empty_result(): 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.hops == []
assert result.completed is False assert result.completed is False
assert result.max_hops_reached is False assert result.max_hops_reached is False