diff --git a/src/netcheck/speed.py b/src/netcheck/speed.py index 39f3608..331f924 100644 --- a/src/netcheck/speed.py +++ b/src/netcheck/speed.py @@ -56,3 +56,158 @@ def bufferbloat_delta(idle_rtt_ms: float | None, loaded_rtts_ms: list[float]) -> return None loaded = percentile(sorted(loaded_rtts_ms), 95.0) return round(max(0.0, loaded - idle_rtt_ms), 3) + + +import asyncio +import time +from typing import Awaitable, Callable + +from netcheck.models import SpeedResult, TierAttempt + +NDT7_CONSENT_NOTICE = ( + "M-Lab NDT7 publishes every measurement as public CC0 open data, including your " + "IP address. Pass --ndt7 only if that is acceptable to you." +) + + +async def run_speed_cascade( + tiers: list[tuple[str, Callable[[], Awaitable[SpeedResult]]]], +) -> SpeedResult: + attempts: list[TierAttempt] = [] + for name, tier in tiers: + began = time.perf_counter() + try: + result = await tier() + except asyncio.CancelledError: + raise + except BaseException as exc: # noqa: BLE001 - a dead tier is data, not control flow + attempts.append( + TierAttempt( + tier=name, + ok=False, + reason=str(exc) or exc.__class__.__name__, + duration_ms=int((time.perf_counter() - began) * 1000), + ) + ) + continue + duration_ms = int((time.perf_counter() - began) * 1000) + if not result.download_mbps: + attempts.append( + TierAttempt(tier=name, ok=False, reason="no throughput measured", duration_ms=duration_ms) + ) + continue + attempts.append(TierAttempt(tier=name, ok=True, reason=None, duration_ms=duration_ms)) + result.tier_attempts = attempts + return result + return SpeedResult(method="none", tier_attempts=attempts) + + +import json + +import httpx + +from netcheck.config import Speedtest + + +async def tier_ookla(binary: str, server: str | None, timeout: float) -> SpeedResult: + args = [binary, "--format=json", "--accept-license", "--accept-gdpr"] + if server: + args += ["--server-id", server] if server.isdigit() else ["--host", server] + process = await asyncio.create_subprocess_exec( + *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL + ) + stdout, _ = await asyncio.wait_for(process.communicate(), timeout=timeout) + payload = json.loads(stdout.decode("utf-8", "replace")) + download = payload.get("download") or {} + upload = payload.get("upload") or {} + server_info = payload.get("server") or {} + return SpeedResult( + method="ookla_bin", + download_mbps=round(mbps(download.get("bytes", 0), (download.get("elapsed", 0) or 0) / 1000), 3), + upload_mbps=round(mbps(upload.get("bytes", 0), (upload.get("elapsed", 0) or 0) / 1000), 3), + server=f"{server_info.get('name', '')} ({server_info.get('location', '')})".strip(), + idle_rtt_ms=(payload.get("ping") or {}).get("latency"), + ) + + +async def tier_cloudflare(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: + down_samples: list[tuple[int, float]] = [] + cfl4: CfL4Stats | None = None + for size in cfg.download_sizes_bytes: + began = time.perf_counter() + response = await client.get( + f"{cfg.cloudflare_base_url}/__down", params={"bytes": size}, timeout=timeout + ) + response.raise_for_status() + down_samples.append((len(response.content), time.perf_counter() - began)) + cfl4 = parse_server_timing_cfl4(response.headers.get("server-timing", "")) or cfl4 + up_samples: list[tuple[int, float]] = [] + for size in cfg.upload_sizes_bytes: + payload = b"\x00" * size + began = time.perf_counter() + response = await client.post(f"{cfg.cloudflare_base_url}/__up", content=payload, timeout=timeout) + response.raise_for_status() + up_samples.append((size, time.perf_counter() - began)) + return SpeedResult( + method="cloudflare", + download_mbps=throughput_from_samples(down_samples), + upload_mbps=throughput_from_samples(up_samples), + server="speed.cloudflare.com", + cfL4_stats=cfl4, + ) + + +async def tier_fastcom(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: + response = await client.get( + cfg.fastcom_api_url, + params={"https": "true", "token": "YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm", "urlCount": "3"}, + timeout=timeout, + ) + response.raise_for_status() + targets = response.json().get("targets") or [] + samples: list[tuple[int, float]] = [] + on_net: bool | None = None + for target in targets: + url = target.get("url") + if not url: + continue + location = (target.get("location") or {}).get("country") + on_net = on_net or bool(location) + began = time.perf_counter() + body = await client.get(url, timeout=timeout) + body.raise_for_status() + samples.append((len(body.content), time.perf_counter() - began)) + return SpeedResult( + method="fastcom", + download_mbps=throughput_from_samples(samples), + upload_mbps=None, + server=targets[0].get("url", "").split("/")[2] if targets else None, + netflix_oca_onnet=on_net, + ) + + +async def tier_ndt7(client: httpx.AsyncClient, cfg: Speedtest, timeout: float) -> SpeedResult: + import websockets + + locate = await client.get(cfg.ndt7_locate_url, timeout=timeout) + locate.raise_for_status() + results = locate.json().get("results") or [] + if not results: + raise RuntimeError("no ndt7 server offered by locate.measurementlab.net") + url = results[0]["urls"]["wss:///ndt/v7/download"] + total = 0 + began = time.perf_counter() + async with websockets.connect(url, subprotocols=["net.measurementlab.ndt.v7"]) as socket: + while time.perf_counter() - began < 10: + try: + message = await asyncio.wait_for(socket.recv(), timeout=timeout) + except (asyncio.TimeoutError, Exception): + break + total += len(message) if isinstance(message, (bytes, bytearray)) else len(message.encode()) + elapsed = time.perf_counter() - began + return SpeedResult( + method="ndt7", + download_mbps=round(mbps(total, elapsed), 3), + upload_mbps=None, + server=results[0].get("machine"), + ) diff --git a/tests/test_speed_math.py b/tests/test_speed_math.py index 6b63ac6..21fbd89 100644 --- a/tests/test_speed_math.py +++ b/tests/test_speed_math.py @@ -84,3 +84,88 @@ def test_bufferbloat_delta_never_goes_negative(): def test_bufferbloat_delta_needs_both_a_baseline_and_samples(): assert bufferbloat_delta(None, [10.0]) is None assert bufferbloat_delta(12.0, []) is None + + +from netcheck.models import SpeedResult, TierAttempt +from netcheck.speed import NDT7_CONSENT_NOTICE, run_speed_cascade + + +async def test_cascade_stops_at_the_first_tier_that_returns_a_download_figure(): + calls: list[str] = [] + + async def ookla() -> SpeedResult: + calls.append("ookla") + return SpeedResult(method="ookla_bin", download_mbps=312.4, upload_mbps=41.0) + + async def cloudflare() -> SpeedResult: + calls.append("cloudflare") + return SpeedResult(method="cloudflare", download_mbps=280.0) + + result = await run_speed_cascade([("ookla_bin", ookla), ("cloudflare", cloudflare)]) + assert result.method == "ookla_bin" + assert result.download_mbps == 312.4 + assert calls == ["ookla"] + assert [a.tier for a in result.tier_attempts] == ["ookla_bin"] + assert result.tier_attempts[0].ok is True + + +async def test_cascade_records_a_failed_tier_and_moves_on(): + async def ookla() -> SpeedResult: + raise FileNotFoundError("speedtest binary not on PATH") + + async def cloudflare() -> SpeedResult: + return SpeedResult(method="cloudflare", download_mbps=280.0) + + result = await run_speed_cascade([("ookla_bin", ookla), ("cloudflare", cloudflare)]) + assert result.method == "cloudflare" + assert [a.tier for a in result.tier_attempts] == ["ookla_bin", "cloudflare"] + assert result.tier_attempts[0].ok is False + assert "not on PATH" in result.tier_attempts[0].reason + assert result.tier_attempts[1].ok is True + + +async def test_a_tier_that_returns_zero_download_counts_as_a_failure(): + async def dead() -> SpeedResult: + return SpeedResult(method="cloudflare", download_mbps=0.0) + + async def alive() -> SpeedResult: + return SpeedResult(method="fastcom", download_mbps=95.0) + + result = await run_speed_cascade([("cloudflare", dead), ("fastcom", alive)]) + assert result.method == "fastcom" + assert result.tier_attempts[0].ok is False + assert result.tier_attempts[0].reason == "no throughput measured" + + +async def test_cascade_exhaustion_is_a_failed_result_not_an_exception(): + async def boom() -> SpeedResult: + raise OSError("network unreachable") + + result = await run_speed_cascade([("ookla_bin", boom), ("cloudflare", boom), ("fastcom", boom)]) + assert isinstance(result, SpeedResult) + assert result.method == "none" + assert result.download_mbps is None + assert [a.tier for a in result.tier_attempts] == ["ookla_bin", "cloudflare", "fastcom"] + assert all(a.ok is False for a in result.tier_attempts) + + +async def test_an_empty_cascade_is_a_failed_result(): + result = await run_speed_cascade([]) + assert result.method == "none" + assert result.tier_attempts == [] + + +async def test_cascade_never_swallows_cancellation(): + import asyncio + + async def cancelled() -> SpeedResult: + raise asyncio.CancelledError() + + with pytest.raises(asyncio.CancelledError): + await run_speed_cascade([("ookla_bin", cancelled)]) + + +def test_the_ndt7_consent_notice_states_what_gets_published(): + assert "CC0" in NDT7_CONSENT_NOTICE + assert "IP" in NDT7_CONSENT_NOTICE + assert "public" in NDT7_CONSENT_NOTICE.lower()