mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 22:42:03 +03:00
66dcd6f5bc
The only feature in this batch adding a real dependency, approved up front and landed the same way tcptrace/ndt7 already do: an optional extra (uv sync --extra quic), imported lazily inside measure_quic() so a missing install degrades to a clear skip warning instead of a crash. Verified the actually-installed aioquic 1.3.0 API directly rather than assuming. connect() takes local_port but no local_host/source-address parameter at all, so --quic structurally cannot honor --interface - _quic_section() says so explicitly rather than silently measuring the OS-default path under a report claiming a forced adapter. Its context manager doesn't yield until the handshake finishes, so unlike the TLS RTT probe's two-connection subtraction trick, handshake time here is timed directly. HandshakeCompleted's fields matched what was expected; dropped a planned quic_version field since aioquic has no clean public API for it rather than reach into a private attribute. quic_verdict(quic_ok, tcp_ok) is the actual payoff: QUIC failing while plain TCP to the same host succeeds means the network specifically drops or throttles UDP/443, silently degrading Chrome/YouTube's HTTP/3 preference in a way no TCP-based diagnostic would ever catch. Both failing is ordinary unreachability already covered elsewhere, so it stays silent rather than duplicating that finding. Opt-in --quic, excluded from --full - an optionally-installed extra must never become a silent requirement. Added the quic extra to the release workflow's wheel-install verification. Live-tested against real cloudflare.com/google.com: real handshakes, ALPN h3 negotiated, correct end-to-end timing. This completes the 16-feature extension batch from plans/zesty-zooming-wilkinson.md (gitignored, local planning doc).
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from netsleuth.config import Thresholds
|
|
from netsleuth.interpret import quic_findings
|
|
from netsleuth.models import QuicResult
|
|
from netsleuth.probes.quic_rtt import quic_verdict
|
|
|
|
|
|
def test_quic_verdict_ok_when_quic_succeeds():
|
|
assert quic_verdict(quic_ok=True, tcp_ok=True) == "ok"
|
|
|
|
|
|
def test_quic_verdict_ok_even_if_tcp_state_is_unknown():
|
|
assert quic_verdict(quic_ok=True, tcp_ok=False) == "ok"
|
|
|
|
|
|
def test_quic_verdict_blocked_when_quic_fails_but_tcp_succeeds():
|
|
assert quic_verdict(quic_ok=False, tcp_ok=True) == "blocked"
|
|
|
|
|
|
def test_quic_verdict_unreachable_when_both_fail():
|
|
assert quic_verdict(quic_ok=False, tcp_ok=False) == "unreachable"
|
|
|
|
|
|
def quic(**kw) -> QuicResult:
|
|
base = dict(label="cloudflare", host="cloudflare.com", port=443)
|
|
base.update(kw)
|
|
return QuicResult(**base)
|
|
|
|
|
|
@pytest.fixture()
|
|
def thresholds() -> Thresholds:
|
|
return Thresholds()
|
|
|
|
|
|
def test_quic_findings_silent_on_a_healthy_result(thresholds):
|
|
assert quic_findings([quic(handshake_ms=50.0, tcp_rtt_ms=20.0)], thresholds) == []
|
|
|
|
|
|
def test_quic_findings_flags_blocked_quic_with_working_tcp(thresholds):
|
|
findings = quic_findings([quic(error="timeout", tcp_rtt_ms=20.0)], thresholds)
|
|
assert [f.id for f in findings] == ["quic.blocked.cloudflare"]
|
|
assert findings[0].severity == "warn"
|
|
|
|
|
|
def test_quic_findings_silent_when_both_quic_and_tcp_fail(thresholds):
|
|
# a fully dead host is latency_findings' job (latency.unreachable), not quic's.
|
|
assert quic_findings([quic(error="timeout", tcp_rtt_ms=None)], thresholds) == []
|
|
|
|
|
|
def test_quic_findings_flags_a_slow_handshake(thresholds):
|
|
findings = quic_findings([quic(handshake_ms=500.0, tcp_rtt_ms=20.0)], thresholds)
|
|
assert [f.id for f in findings] == ["quic.handshake_slow.cloudflare"]
|
|
|
|
|
|
def test_quic_findings_fast_handshake_produces_nothing(thresholds):
|
|
findings = quic_findings([quic(handshake_ms=50.0, tcp_rtt_ms=20.0)], thresholds)
|
|
assert findings == []
|
|
|
|
|
|
aioquic = pytest.importorskip("aioquic")
|
|
|
|
|
|
async def test_measure_quic_reports_an_error_for_an_unreachable_host():
|
|
from netsleuth.probes.quic_rtt import measure_quic
|
|
|
|
result = await measure_quic("dead", "192.0.2.1", port=1, timeout=0.5)
|
|
assert result.label == "dead"
|
|
assert result.error
|
|
assert result.handshake_ms is None
|
|
|
|
|
|
async def test_quic_fanout_returns_a_result_per_target():
|
|
from netsleuth.probes.quic_rtt import quic_fanout
|
|
|
|
results = await quic_fanout([("a", "192.0.2.1"), ("b", "192.0.2.2")], port=1, timeout=0.5, concurrency=2)
|
|
assert len(results) == 2
|
|
assert all(r.error for r in results)
|