netsleuth/tests/test_interpret.py

from __future__ import annotations

import pytest

from netsleuth.config import Band, Thresholds
from netsleuth.models import PingResult, TraceHop, TraceResult
from netsleuth.interpret import dual_family_findings, latency_findings, path_findings, severity_for


@pytest.fixture()
def thresholds() -> Thresholds:
    return Thresholds()


def test_severity_bands_are_inclusive_at_the_good_edge():
    band = Band(good=40.0, warn=100.0)
    assert severity_for(10.0, band) == "ok"
    assert severity_for(40.0, band) == "ok"
    assert severity_for(40.1, band) == "warn"
    assert severity_for(100.0, band) == "warn"
    assert severity_for(100.1, band) == "crit"


def test_severity_of_a_missing_value_is_info_not_ok():
    assert severity_for(None, Band(good=40.0, warn=100.0)) == "info"


def test_severity_can_be_inverted_for_metrics_where_higher_is_better():
    band = Band(good=50.0, warn=10.0)
    assert severity_for(100.0, band, higher_is_worse=False) == "ok"
    assert severity_for(50.0, band, higher_is_worse=False) == "ok"
    assert severity_for(30.0, band, higher_is_worse=False) == "warn"
    assert severity_for(5.0, band, higher_is_worse=False) == "crit"


def ping(**kw) -> PingResult:
    base = dict(
        label="cloudflare-dns",
        host="1.1.1.1",
        resolved_ip="1.1.1.1",
        method="icmp_dgram",
        sent=20,
        received=20,
        loss_pct=0.0,
        min_ms=10.0,
        avg_ms=12.0,
        max_ms=15.0,
        mdev_ms=1.0,
        jitter_ms=1.0,
        samples=[],
    )
    base.update(kw)
    return PingResult(**base)


def test_a_healthy_ping_produces_no_findings(thresholds):
    assert latency_findings([ping()], thresholds) == []


def test_high_latency_produces_a_warn_finding_with_metric_and_threshold(thresholds):
    findings = latency_findings([ping(avg_ms=130.0)], thresholds)
    assert len(findings) == 1
    f = findings[0]
    assert f.id == "latency.avg.cloudflare-dns"
    assert f.severity == "crit"
    assert f.metric == "avg_ms"
    assert f.value == 130.0
    assert f.threshold == 100.0
    assert f.advice


def test_high_jitter_and_loss_each_produce_their_own_finding(thresholds):
    findings = latency_findings([ping(jitter_ms=25.0, loss_pct=5.0, received=19)], thresholds)
    ids = sorted(f.id for f in findings)
    assert ids == ["latency.jitter.cloudflare-dns", "latency.loss.cloudflare-dns"]


def test_tcp_measured_loss_is_labelled_as_connection_failures_not_packet_loss(thresholds):
    findings = latency_findings([ping(method="tcp", loss_pct=10.0)], thresholds)
    loss = [f for f in findings if f.id.startswith("latency.loss")][0]
    assert "connection failure" in loss.detail.lower()
    assert "packet loss" not in loss.detail.lower()


def test_a_fully_dead_host_is_critical_regardless_of_bands(thresholds):
    findings = latency_findings(
        [ping(received=0, loss_pct=100.0, avg_ms=None, min_ms=None, max_ms=None, jitter_ms=None)],
        thresholds,
    )
    assert [f.severity for f in findings] == ["crit"]
    assert findings[0].id == "latency.unreachable.cloudflare-dns"


def test_dual_family_findings_silent_when_no_v6_hosts_present():
    assert dual_family_findings([ping()]) == []


def test_dual_family_findings_reports_unreachable_when_every_v6_host_is_dead():
    pings = [
        ping(),
        ping(label="cloudflare-dns-v6", host="2606:4700:4700::1111", received=0, avg_ms=None, loss_pct=100.0),
    ]
    findings = dual_family_findings(pings)
    assert [f.id for f in findings] == ["net.ipv6_unreachable"]
    assert findings[0].severity == "info"


def test_dual_family_findings_flags_a_much_slower_v6_path():
    pings = [
        ping(avg_ms=10.0),
        ping(label="cloudflare-dns-v6", host="2606:4700:4700::1111", avg_ms=80.0),
    ]
    findings = dual_family_findings(pings)
    assert [f.id for f in findings] == ["latency.v6_much_slower.cloudflare-dns"]
    assert findings[0].severity == "warn"


def test_dual_family_findings_silent_when_v6_is_only_slightly_slower():
    pings = [
        ping(avg_ms=10.0),
        ping(label="cloudflare-dns-v6", host="2606:4700:4700::1111", avg_ms=15.0),
    ]
    assert dual_family_findings(pings) == []


def test_dual_family_findings_silent_when_there_is_no_v4_twin():
    pings = [ping(label="cloudflare-dns-v6", host="2606:4700:4700::1111", avg_ms=80.0)]
    assert dual_family_findings(pings) == []


def test_path_findings_highlight_the_first_sustained_loss_jump():
    trace = TraceResult(
        target="1.1.1.1",
        resolved_ip="1.1.1.1",
        backend="mtr_json",
        hops=[
            TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0),
            TraceHop(ttl=2, ip="10.64.0.1", loss_pct=0.0, avg_ms=9.0),
            TraceHop(ttl=3, ip="198.51.100.7", loss_pct=60.0, avg_ms=40.0),
            TraceHop(ttl=4, ip="1.1.1.1", loss_pct=55.0, avg_ms=41.0),
        ],
        completed=True,
    )
    findings = path_findings(trace)
    loss = [f for f in findings if f.id == "path.loss_jump"][0]
    assert loss.severity == "crit"
    assert "hop 3" in loss.detail
    assert "198.51.100.7" in loss.detail


def test_path_findings_label_a_non_responding_hop_instead_of_printing_none():
    trace = TraceResult(
        target="1.1.1.1",
        resolved_ip="1.1.1.1",
        backend="mtr_json",
        hops=[
            TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0),
            TraceHop(ttl=2, ip=None, loss_pct=100.0, avg_ms=None),
            TraceHop(ttl=3, ip="1.1.1.1", loss_pct=100.0, avg_ms=None),
        ],
        completed=True,
    )
    findings = path_findings(trace)
    loss = [f for f in findings if f.id == "path.loss_jump"][0]
    assert "None" not in loss.detail
    assert "no reply" in loss.detail


def test_path_findings_ignore_a_single_hop_that_rate_limits_icmp():
    trace = TraceResult(
        target="1.1.1.1",
        resolved_ip="1.1.1.1",
        backend="mtr_json",
        hops=[
            TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0),
            TraceHop(ttl=2, ip="10.64.0.1", loss_pct=100.0, avg_ms=None),
            TraceHop(ttl=3, ip="1.1.1.1", loss_pct=0.0, avg_ms=12.0),
        ],
        completed=True,
    )
    assert [f.id for f in path_findings(trace)] == []


def test_path_findings_report_an_incomplete_trace():
    trace = TraceResult(target="1.1.1.1", resolved_ip="1.1.1.1", backend="icmplib", hops=[], completed=False)
    assert [f.id for f in path_findings(trace)] == ["path.incomplete"]


def test_path_findings_ignore_a_long_run_of_silent_hops_that_recovers_at_the_destination():
    # Real transit routers commonly rate-limit ICMP for many consecutive hops in a
    # row (not just one) while the actual destination still answers cleanly.
    trace = TraceResult(
        target="8.8.8.8",
        resolved_ip="8.8.8.8",
        backend="icmp_win",
        hops=[TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0)]
        + [TraceHop(ttl=n, ip=None, loss_pct=100.0, avg_ms=None) for n in range(2, 11)]
        + [TraceHop(ttl=11, ip="8.8.8.8", loss_pct=0.0, avg_ms=7.0)],
        completed=True,
    )
    assert [f.id for f in path_findings(trace)] == []


def test_path_findings_flag_loss_that_persists_all_the_way_to_the_last_hop():
    trace = TraceResult(
        target="8.8.8.8",
        resolved_ip="8.8.8.8",
        backend="icmp_win",
        hops=[TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0)]
        + [TraceHop(ttl=n, ip=None, loss_pct=100.0, avg_ms=None) for n in range(2, 11)],
        completed=False,
    )
    findings = path_findings(trace)
    loss = [f for f in findings if f.id == "path.loss_jump"]
    assert len(loss) == 1
    assert "hop 2" in loss[0].detail


from netsleuth.config import VpnBands
from netsleuth.models import AdapterLeakResult, CfTrace, DnsLeak, IpGeo, LocalNet, Signal, VpnContext
from netsleuth.interpret import SIGNAL_WEIGHTS, assess_vpn, gather_vpn_signals, score_vpn


def sig(name: str, observed: bool = True) -> Signal:
    return Signal(name=name, observed=observed, weight=SIGNAL_WEIGHTS[name], direction="vpn")


def test_no_signals_means_no_vpn():
    verdict, confidence = score_vpn([], VpnBands())
    assert verdict == "none"
    assert confidence == 0.0


def test_unobserved_signals_contribute_nothing():
    verdict, confidence = score_vpn(
        [sig("tunnel_iface", observed=False), sig("cf_warp", observed=False)], VpnBands()
    )
    assert verdict == "none"
    assert confidence == 0.0


def test_cloudflare_warp_alone_is_enough_for_likely():
    verdict, confidence = score_vpn([sig("cf_warp")], VpnBands())
    assert verdict == "likely"
    assert confidence == pytest.approx(0.50)


def test_tunnel_interface_plus_hosting_egress_is_confirmed():
    verdict, confidence = score_vpn([sig("tunnel_iface"), sig("provider_hosting")], VpnBands())
    assert verdict == "confirmed"
    assert confidence >= 0.75


def test_a_mobile_flag_with_a_timezone_mismatch_does_not_fire():
    # This combination is normal for anyone travelling on a phone hotspot and
    # must not be reported as a VPN.
    verdict, confidence = score_vpn([sig("provider_mobile"), sig("timezone_mismatch")], VpnBands())
    assert verdict == "none"
    assert confidence < 0.40


def test_mtu_anomaly_alone_does_not_fire():
    verdict, _ = score_vpn([sig("mtu_anomaly")], VpnBands())
    assert verdict == "none"


def test_dns_asn_mismatch_alone_does_not_fire():
    verdict, _ = score_vpn([sig("dns_asn_mismatch")], VpnBands())
    assert verdict == "none"


def test_mtu_anomaly_plus_tunnel_iface_plus_dns_mismatch_is_confirmed():
    verdict, confidence = score_vpn(
        [sig("tunnel_iface"), sig("mtu_anomaly"), sig("dns_asn_mismatch")], VpnBands()
    )
    assert verdict == "confirmed"
    assert confidence == pytest.approx(0.80)


def test_confidence_is_capped_at_one():
    _, confidence = score_vpn([sig(name) for name in SIGNAL_WEIGHTS], VpnBands())
    assert confidence == 1.0


def test_clean_direction_signals_reduce_confidence():
    signals = [
        sig("provider_hosting"),
        Signal(name="pdb_eyeball_isp", observed=True, weight=0.2, direction="clean"),
    ]
    _, confidence = score_vpn(signals, VpnBands())
    assert confidence == pytest.approx(0.20)


def test_gather_signals_flags_a_wireguard_tunnel_with_a_hosting_egress():
    local = LocalNet(iface_name="wg0", local_ipv4="10.7.0.2", iface_mtu=1420, default_gateway_v4="10.7.0.1")
    geo = IpGeo(ip="203.0.113.44", asn="AS64500", country_code="NL", timezone="Europe/Amsterdam")
    ctx = VpnContext(
        local=local,
        geo=geo,
        cf=CfTrace(ip="203.0.113.44", warp="off"),
        dns_leak=None,
        pdb_info_type="NSP",
        os_timezone="Europe/Amsterdam",
        provider_flags={"hosting": True, "proxy": False, "mobile": False},
    )
    signals = {s.name: s for s in gather_vpn_signals(ctx)}
    assert signals["tunnel_iface"].observed is True
    assert signals["tunnel_iface"].note == "wg0"
    assert signals["mtu_anomaly"].observed is True
    assert signals["mtu_anomaly"].note == "wireguard"
    assert signals["provider_hosting"].observed is True
    assert signals["cf_warp"].observed is False
    assert signals["timezone_mismatch"].observed is False


def test_gather_signals_detects_a_dns_resolver_in_another_asn():
    leak = DnsLeak(
        per_adapter=[
            AdapterLeakResult(
                adapter="Wi-Fi",
                configured_resolvers=["192.168.1.1"],
                echoed_ip="203.0.113.9",
                echoed_asn="AS64501",
                matches_egress_asn=False,
            )
        ]
    )
    ctx = VpnContext(
        local=LocalNet(iface_name="eth0"),
        geo=IpGeo(asn="AS64500"),
        cf=None,
        dns_leak=leak,
        pdb_info_type=None,
        os_timezone=None,
        provider_flags={},
    )
    signals = {s.name: s for s in gather_vpn_signals(ctx)}
    assert signals["dns_asn_mismatch"].observed is True
    assert "Wi-Fi" in signals["dns_asn_mismatch"].note


def test_gather_signals_detects_cloudflare_warp():
    ctx = VpnContext(
        local=LocalNet(),
        geo=IpGeo(),
        cf=CfTrace(warp="on"),
        dns_leak=None,
        pdb_info_type=None,
        os_timezone=None,
        provider_flags={},
    )
    signals = {s.name: s for s in gather_vpn_signals(ctx)}
    assert signals["cf_warp"].observed is True


def test_assess_vpn_returns_a_complete_assessment():
    assessment = assess_vpn(
        signals=[sig("tunnel_iface"), sig("provider_hosting")],
        bands=VpnBands(),
        tunnel_iface="wg0",
        dns_leak=None,
    )
    assert assessment.verdict == "confirmed"
    assert assessment.tunnel_iface == "wg0"
    assert len(assessment.signals) == 2
    assert assessment.confidence >= 0.75


from netsleuth.config import BufferbloatBands
from netsleuth.models import Finding, SpeedResult
from netsleuth.interpret import bufferbloat_consequence, grade_bufferbloat, overall_verdict, speed_findings


def test_bufferbloat_grades_follow_the_configured_bands():
    b = BufferbloatBands()
    assert grade_bufferbloat(0.0, b) == "A"
    assert grade_bufferbloat(5.0, b) == "A"
    assert grade_bufferbloat(5.1, b) == "B"
    assert grade_bufferbloat(30.0, b) == "B"
    assert grade_bufferbloat(60.0, b) == "C"
    assert grade_bufferbloat(200.0, b) == "D"
    assert grade_bufferbloat(400.0, b) == "E"
    assert grade_bufferbloat(400.1, b) == "F"


def test_bufferbloat_grade_of_an_unmeasured_delta_is_unknown():
    assert grade_bufferbloat(None, BufferbloatBands()) == "?"


def test_negative_delta_is_graded_a_not_crashed():
    assert grade_bufferbloat(-3.0, BufferbloatBands()) == "A"


def test_each_grade_has_a_plain_language_consequence():
    for grade in "ABCDEF?":
        assert bufferbloat_consequence(grade)
    assert "call" in bufferbloat_consequence("F").lower()


def test_speed_findings_report_a_bad_bufferbloat_grade():
    speed = SpeedResult(method="cloudflare", download_mbps=300.0, upload_mbps=40.0, bufferbloat_down_ms=150.0)
    findings = speed_findings(speed, BufferbloatBands())
    bloat = [f for f in findings if f.id == "speed.bufferbloat_down"][0]
    assert bloat.severity == "crit"
    assert bloat.value == "D"
    assert "call" in bloat.advice.lower() or "queue" in bloat.advice.lower()


def test_speed_findings_are_silent_on_a_clean_line():
    speed = SpeedResult(
        method="ookla_bin",
        download_mbps=300.0,
        upload_mbps=40.0,
        bufferbloat_down_ms=3.0,
        bufferbloat_up_ms=4.0,
    )
    assert speed_findings(speed, BufferbloatBands()) == []


def test_speed_findings_report_an_exhausted_cascade():
    speed = SpeedResult(method="none")
    findings = speed_findings(speed, BufferbloatBands())
    assert [f.id for f in findings] == ["speed.unavailable"]
    assert findings[0].severity == "info"


def test_speed_findings_flags_a_speedtest_server_in_another_country():
    speed = SpeedResult(method="ookla_bin", download_mbps=100.0, upload_mbps=20.0, server_country="Germany")
    findings = speed_findings(speed, BufferbloatBands(), client_country="Russia")
    assert [f.id for f in findings] == ["speed.server_other_country"]
    assert findings[0].severity == "info"


def test_speed_findings_silent_when_server_country_matches_client():
    speed = SpeedResult(method="ookla_bin", download_mbps=100.0, upload_mbps=20.0, server_country="Germany")
    assert speed_findings(speed, BufferbloatBands(), client_country="Germany") == []


def test_speed_findings_silent_when_either_country_is_unknown():
    speed = SpeedResult(method="ookla_bin", download_mbps=100.0, upload_mbps=20.0, server_country=None)
    assert speed_findings(speed, BufferbloatBands(), client_country="Russia") == []
    assert speed_findings(speed, BufferbloatBands(), client_country=None) == []


def test_overall_verdict_is_healthy_with_no_findings():
    status, score, summary = overall_verdict([])
    assert status == "ok"
    assert score == 100
    assert "no problems" in summary.lower()


def test_overall_verdict_degrades_with_severity():
    warn = Finding(id="a", severity="warn", title="Jitter high", detail="")
    crit = Finding(id="b", severity="crit", title="Loss on path", detail="")
    info = Finding(id="c", severity="info", title="No speed data", detail="")

    status, score, summary = overall_verdict([info])
    assert (status, score) == ("ok", 97)

    status, score, summary = overall_verdict([warn])
    assert status == "warn"
    assert score == 90

    status, score, summary = overall_verdict([warn, crit])
    assert status == "crit"
    assert score == 65
    assert "Loss on path" in summary


def test_overall_score_never_goes_below_zero():
    findings = [Finding(id=f"f{i}", severity="crit", title="x", detail="") for i in range(20)]
    status, score, _ = overall_verdict(findings)
    assert status == "crit"
    assert score == 0


_CYRILLIC = set("абвгдеёжзийклмнопрстуфхцчшщъыьэюя")


def has_cyrillic(text: str) -> bool:
    return any(ch in _CYRILLIC for ch in text.lower())


def test_latency_findings_carry_russian_siblings():
    findings = latency_findings(
        [
            ping(received=0, loss_pct=100.0, avg_ms=None, min_ms=None, max_ms=None, jitter_ms=None),
            ping(avg_ms=130.0, jitter_ms=25.0, loss_pct=5.0, received=19),
        ],
        thresholds_fixture(),
    )
    assert findings
    for f in findings:
        assert f.title_ru and has_cyrillic(f.title_ru)
        assert f.detail_ru and has_cyrillic(f.detail_ru)
        if f.advice:
            assert f.advice_ru and has_cyrillic(f.advice_ru)


def test_path_findings_carry_russian_siblings_for_every_case():
    no_hops = TraceResult(target="1.1.1.1", resolved_ip="1.1.1.1", backend="icmplib", hops=[])
    findings = path_findings(no_hops)
    assert findings[0].title_ru and has_cyrillic(findings[0].title_ru)
    assert findings[0].detail_ru and has_cyrillic(findings[0].detail_ru)
    assert findings[0].advice_ru and has_cyrillic(findings[0].advice_ru)

    loss_jump_trace = TraceResult(
        target="1.1.1.1",
        resolved_ip="1.1.1.1",
        backend="mtr_json",
        hops=[
            TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0),
            TraceHop(ttl=2, ip="10.64.0.1", loss_pct=0.0, avg_ms=9.0),
            TraceHop(ttl=3, ip="198.51.100.7", loss_pct=60.0, avg_ms=40.0),
            TraceHop(ttl=4, ip="1.1.1.1", loss_pct=55.0, avg_ms=41.0),
        ],
        completed=True,
    )
    loss_findings = path_findings(loss_jump_trace)
    loss = [f for f in loss_findings if f.id == "path.loss_jump"][0]
    assert loss.title_ru and has_cyrillic(loss.title_ru)
    assert loss.detail_ru and has_cyrillic(loss.detail_ru)
    assert loss.advice_ru and has_cyrillic(loss.advice_ru)

    incomplete_trace = TraceResult(
        target="1.1.1.1",
        resolved_ip="1.1.1.1",
        backend="mtr_json",
        hops=[TraceHop(ttl=1, ip="192.168.1.1", loss_pct=0.0, avg_ms=1.0)],
        completed=False,
    )
    incomplete = path_findings(incomplete_trace)
    incomplete_finding = [f for f in incomplete if f.id == "path.incomplete"][0]
    assert incomplete_finding.title_ru and has_cyrillic(incomplete_finding.title_ru)
    assert incomplete_finding.detail_ru and has_cyrillic(incomplete_finding.detail_ru)
    assert incomplete_finding.advice_ru and has_cyrillic(incomplete_finding.advice_ru)


def thresholds_fixture() -> Thresholds:
    return Thresholds()


def test_speed_findings_carry_russian_siblings():
    unavailable = speed_findings(SpeedResult(method="none"), BufferbloatBands())
    assert unavailable[0].title_ru and has_cyrillic(unavailable[0].title_ru)
    assert unavailable[0].detail_ru and has_cyrillic(unavailable[0].detail_ru)
    assert unavailable[0].advice_ru and has_cyrillic(unavailable[0].advice_ru)

    speed = SpeedResult(
        method="cloudflare",
        download_mbps=300.0,
        upload_mbps=40.0,
        bufferbloat_down_ms=150.0,
        bufferbloat_up_ms=150.0,
    )
    findings = speed_findings(speed, BufferbloatBands())
    assert len(findings) == 2
    for f in findings:
        assert f.title_ru and has_cyrillic(f.title_ru)
        assert f.detail_ru and has_cyrillic(f.detail_ru)
        assert f.advice_ru and has_cyrillic(f.advice_ru)
    down = [f for f in findings if f.id == "speed.bufferbloat_down"][0]
    up = [f for f in findings if f.id == "speed.bufferbloat_up"][0]
    assert "приём" in down.title_ru
    assert "передача" in up.title_ru


def test_bufferbloat_consequence_russian_covers_every_grade():
    for grade in "ABCDEF?":
        text = bufferbloat_consequence(grade, lang="ru")
        assert text and has_cyrillic(text)


def test_bufferbloat_consequence_default_english_is_unchanged():
    for grade in "ABCDEF?":
        assert bufferbloat_consequence(grade) == bufferbloat_consequence(grade, lang="en")
    assert "call" in bufferbloat_consequence("F").lower()