diff --git a/src/netcheck/traceparse.py b/src/netcheck/traceparse.py index 9837195..5df9f76 100644 --- a/src/netcheck/traceparse.py +++ b/src/netcheck/traceparse.py @@ -71,3 +71,42 @@ def parse_linux(text: str) -> list[TraceHop]: ) hops.append(finalize_hop(hop)) return hops + + +# tracert reports any sub-millisecond RTT as "<1 ms"; the midpoint of (0, 1) is +# the least-wrong single value to record for it. +WINDOWS_SUB_MS = 0.5 + +_WIN_PROBE_RE = re.compile(r"(?P<)?\s*(?P\d+)\s*ms|(?P\*)") +_WIN_BRACKET_RE = re.compile(r"\[(?P[0-9a-fA-F:.%]+)\]") +_WIN_NAME_RE = re.compile(r"(?P[A-Za-z0-9._-]+)\s*\[") + + +def parse_windows(text: str) -> list[TraceHop]: + hops: list[TraceHop] = [] + for line in text.splitlines(): + match = _HOP_LINE_RE.match(line) + if not match: + continue + ttl, body = int(match.group(1)), match.group(2) + probes: list[float | None] = [] + tail_start = 0 + for probe in _WIN_PROBE_RE.finditer(body): + if probe.group("star"): + probes.append(None) + else: + probes.append(WINDOWS_SUB_MS if probe.group("lt") else float(probe.group("rtt"))) + tail_start = probe.end() + tail = body[tail_start:] + bracket = _WIN_BRACKET_RE.search(tail) + if bracket: + ip = bracket.group("ip") + name_match = _WIN_NAME_RE.search(tail) + rdns = name_match.group("name") if name_match else None + else: + ip = _extract_ip(tail) + rdns = None + if rdns == ip: + rdns = None + hops.append(finalize_hop(TraceHop(ttl=ttl, ip=ip, reverse_dns=rdns, probes=probes))) + return hops diff --git a/tests/fixtures/traceroute/windows/tracert_en.txt b/tests/fixtures/traceroute/windows/tracert_en.txt new file mode 100644 index 0000000..ab3c198 --- /dev/null +++ b/tests/fixtures/traceroute/windows/tracert_en.txt @@ -0,0 +1,11 @@ + +Tracing route to one.one.one.one [1.1.1.1] +over a maximum of 30 hops: + + 1 1 ms <1 ms 1 ms 192.168.1.1 + 2 9 ms 8 ms 9 ms 10.64.0.1 + 3 * * * Request timed out. + 4 12 ms 12 ms 11 ms ae-1.r01.ams.example.net [77.88.1.5] + 5 14 ms * 13 ms one.one.one.one [1.1.1.1] + +Trace complete. diff --git a/tests/fixtures/traceroute/windows/tracert_ru_cp866.txt b/tests/fixtures/traceroute/windows/tracert_ru_cp866.txt new file mode 100644 index 0000000..b0cba29 --- /dev/null +++ b/tests/fixtures/traceroute/windows/tracert_ru_cp866.txt @@ -0,0 +1,11 @@ + +Трассировка маршрута к one.one.one.one [1.1.1.1] +с максимальным числом прыжков 30: + + 1 1 ms <1 ms 1 ms 192.168.1.1 + 2 9 ms 8 ms 9 ms 10.64.0.1 + 3 * * * Превышен интервал ожидания для запроса. + 4 12 ms 12 ms 11 ms ae-1.r01.ams.example.net [77.88.1.5] + 5 14 ms * 13 ms one.one.one.one [1.1.1.1] + +Трассировка завершена. diff --git a/tests/fixtures/traceroute/windows/tracert_unreachable_en.txt b/tests/fixtures/traceroute/windows/tracert_unreachable_en.txt new file mode 100644 index 0000000..7b63eea --- /dev/null +++ b/tests/fixtures/traceroute/windows/tracert_unreachable_en.txt @@ -0,0 +1,8 @@ + +Tracing route to 203.0.113.9 over a maximum of 30 hops + + 1 1 ms <1 ms <1 ms 192.168.1.1 + 2 10 ms 9 ms 9 ms 10.64.0.1 + 3 border.example.net [198.51.100.7] reports: Destination net unreachable. + +Trace complete. diff --git a/tests/test_traceparse.py b/tests/test_traceparse.py index d104789..39908f4 100644 --- a/tests/test_traceparse.py +++ b/tests/test_traceparse.py @@ -71,3 +71,73 @@ def test_linux_max_hops_run_yields_trailing_dead_hops(trace_fixture): def test_linux_parser_tolerates_empty_and_garbage_input(): assert parse_linux("") == [] assert parse_linux("bash: traceroute: command not found\n") == [] + + +from netcheck.traceparse import WINDOWS_SUB_MS, parse_windows + + +@pytest.fixture() +def win_en(trace_fixture): + return parse_windows(trace_fixture("windows", "tracert_en.txt")) + + +def test_windows_hop_numbers_and_count(win_en): + assert [hop.ttl for hop in win_en] == [1, 2, 3, 4, 5] + + +def test_windows_sub_millisecond_probe_is_not_dropped(win_en): + assert win_en[0].probes == [1.0, WINDOWS_SUB_MS, 1.0] + assert 0.0 < WINDOWS_SUB_MS < 1.0 + + +def test_windows_bracketed_ip_and_hostname_are_split(win_en): + assert win_en[3].ip == "77.88.1.5" + assert win_en[3].reverse_dns == "ae-1.r01.ams.example.net" + + +def test_windows_bare_ip_hop_has_no_reverse_dns(win_en): + assert win_en[1].ip == "10.64.0.1" + assert win_en[1].reverse_dns is None + + +def test_windows_timed_out_hop_is_detected_by_shape_not_by_message(win_en): + hop = win_en[2] + assert hop.probes == [None, None, None] + assert hop.ip is None + assert hop.loss_pct == 100.0 + + +def test_windows_partial_timeout_keeps_probe_order(win_en): + assert win_en[4].probes == [14.0, None, 13.0] + assert win_en[4].ip == "1.1.1.1" + + +def test_windows_headers_and_footer_are_not_parsed_as_hops(win_en): + assert all(hop.ttl <= 5 for hop in win_en) + assert len(win_en) == 5 + + +def test_windows_cp866_localized_output_parses_identically(trace_fixture): + ru = parse_windows(trace_fixture("windows", "tracert_ru_cp866.txt", encoding="cp866")) + en = parse_windows(trace_fixture("windows", "tracert_en.txt")) + assert [h.ttl for h in ru] == [h.ttl for h in en] + assert [h.ip for h in ru] == [h.ip for h in en] + assert [h.probes for h in ru] == [h.probes for h in en] + + +def test_windows_cp866_timed_out_hop_has_no_spurious_ip(trace_fixture): + ru = parse_windows(trace_fixture("windows", "tracert_ru_cp866.txt", encoding="cp866")) + assert ru[2].ip is None + assert ru[2].probes == [None, None, None] + + +def test_windows_destination_unreachable_hop_keeps_its_ip(trace_fixture): + hops = parse_windows(trace_fixture("windows", "tracert_unreachable_en.txt")) + assert hops[-1].ttl == 3 + assert hops[-1].ip == "198.51.100.7" + assert hops[-1].probes == [] + assert hops[-1].loss_pct == 0.0 + + +def test_windows_parser_tolerates_empty_input(): + assert parse_windows("") == []