Blame for tests/test_speed.py

import asyncio
import json
import os
import shutil
import stat
import sys
import time

import pytest

from bot import speed
from bot.speed import SpeedError, SpeedResult

CLI_RESULT = {
    "type": "result", "timestamp": "2026-10-21T04:00:00Z",
    "ping": {"jitter": 0.52, "latency": 4.21, "low": 3.9, "high": 5.0},
    "download": {"bandwidth": 31_250_000, "bytes": 400_000_000, "elapsed": 12000},
    "upload": {"bandwidth": 12_500_000, "bytes": 150_000_000, "elapsed": 11000},
    "packetLoss": 0, "isp": "Hetzner Online GmbH",
    "server": {"id": 1234, "host": "speed.example.net", "name": "Example Telecom", "location": "Frankfurt", "country": "Germany"},
    "result": {"id": "abc", "url": "https://www.speedtest.net/result/c/abc"},
}
PY_RESULT = {
    "download": 245_310_000.0, "upload": 118_200_000.0, "ping": 11.5, "share": "http://www.speedtest.net/result/1.png",
    "server": {"sponsor": "Example Telecom", "name": "Frankfurt", "country": "Germany"},
    "client": {"isp": "Hetzner Online GmbH", "ip": "1.2.3.4"},
}


def fake_binary(directory, name, version_text, body):
    """An executable `name` that answers --version with ``version_text`` and anything else by running ``body`` (sh)."""
    path = directory / name
    path.write_text(f'#!/bin/sh\nif [ "$1" = "--version" ]; then\n  echo "{version_text}"\n  exit 0\nfi\n{body}\n')
    path.chmod(path.stat().st_mode | stat.S_IEXEC)
    return path


SLEEP = shutil.which("sleep") or "/bin/sleep"  # PATH is replaced in these tests, so scripts need absolute paths
CAT = shutil.which("cat") or "/bin/cat"


def emit(bindir, obj):
    """Shell line that prints ``obj`` as one JSON line (it lives in a file, so quotes inside cannot break the script)."""
    path = bindir.parent / f"out{len(list(bindir.parent.glob('out*.json')))}.json"
    path.write_text(json.dumps(obj) + "\n")
    return f"{CAT} {path}"


@pytest.fixture
def bindir(tmp_path, monkeypatch):
    d = tmp_path / "bin"
    d.mkdir()
    monkeypatch.setenv("PATH", str(d))  # only our fakes can be found
    monkeypatch.delenv("SPEEDTEST_BIN", raising=False)
    return d


# ---------------------------------------------------------------------------
# parsing
# ---------------------------------------------------------------------------


def test_parse_cli_converts_bytes_per_second_to_megabits():
    r = speed.parse_cli(CLI_RESULT)
    assert round(r.download_mbps, 2) == 250.0 and round(r.upload_mbps, 2) == 100.0
    assert r.ping_ms == 4.21 and r.jitter_ms == 0.52 and r.packet_loss == 0
    assert r.isp == "Hetzner Online GmbH" and r.server == "Example Telecom" and r.location == "Frankfurt, Germany"
    assert r.url == "https://www.speedtest.net/result/c/abc" and r.source == "Ookla Speedtest CLI"


def test_parse_cli_copes_with_missing_details_but_not_missing_speeds():
    r = speed.parse_cli({"download": {"bandwidth": 1_000_000}, "upload": {"bandwidth": 500_000}})
    assert r.ping_ms == 0.0 and r.jitter_ms is None and r.packet_loss is None and r.isp == "" and r.url == "" and r.location == ""
    for broken in ({}, {"download": {"bandwidth": 1}}, {"download": {"bandwidth": "x"}, "upload": {"bandwidth": 1}}):
        with pytest.raises(SpeedError):
            speed.parse_cli(broken)


def test_parse_python_converts_bits_per_second_to_megabits():
    r = speed.parse_python(PY_RESULT)
    assert round(r.download_mbps, 2) == 245.31 and round(r.upload_mbps, 2) == 118.2 and r.ping_ms == 11.5
    assert r.isp == "Hetzner Online GmbH" and r.server == "Example Telecom" and r.location == "Frankfurt, Germany"
    assert r.url.endswith("1.png") and "speedtest-cli" in r.source
    with pytest.raises(SpeedError):
        speed.parse_python({"ping": 3})


def test_the_result_line_is_picked_out_of_the_cli_output():
    out = "\n".join(["noise", json.dumps({"type": "log", "message": "x"}), json.dumps(CLI_RESULT), "", "{broken json"])
    assert speed._last_result(out)["isp"] == "Hetzner Online GmbH"
    assert speed._last_result("nothing useful") is None
    assert speed._last_result(json.dumps({"type": "log", "message": "x"})) is None


def test_the_reason_for_a_failure_comes_from_the_log_line():
    out = json.dumps({"type": "log", "level": "error", "message": "Configuration - Couldn't resolve host name"})
    assert speed._log_message(out, "") == "Configuration - Couldn't resolve host name"
    assert speed._log_message("", "segmentation fault") == "segmentation fault"
    assert speed._log_message("", "") == "no output"


# ---------------------------------------------------------------------------
# finding and running the CLI
# ---------------------------------------------------------------------------


async def test_the_official_cli_is_found_by_its_version_text(bindir):
    assert await speed.find_ookla_cli() is None
    fake_binary(bindir, "speedtest", "speedtest-cli 2.1.3\nPython 3.12", "exit 1")  # the pip package's script is not the CLI
    assert await speed.find_ookla_cli() is None
    ookla = fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84 (5ae238b) Linux/x86_64", "exit 1")
    assert await speed.find_ookla_cli() == str(ookla)


async def test_a_plain_speedtest_binary_is_accepted_when_it_is_the_ookla_one(bindir):
    path = fake_binary(bindir, "speedtest", "Speedtest by Ookla 1.2.0.84", "exit 1")
    assert await speed.find_ookla_cli() == str(path)


async def test_the_speedtest_bin_variable_wins(bindir, tmp_path, monkeypatch):
    other = tmp_path / "elsewhere"
    other.mkdir()
    chosen = fake_binary(other, "mytest", "Speedtest by Ookla 1.2.0.84", "exit 1")
    fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", "exit 1")
    monkeypatch.setenv("SPEEDTEST_BIN", str(chosen))
    assert await speed.find_ookla_cli() == str(chosen)


async def test_run_cli_returns_the_parsed_result(bindir):
    path = fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", emit(bindir, CLI_RESULT))
    r = await speed.run_cli(str(path))
    assert round(r.download_mbps) == 250 and r.server == "Example Telecom"


async def test_run_cli_passes_the_flags_that_make_it_non_interactive(bindir, tmp_path):
    args_file = tmp_path / "args.txt"
    path = fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", f'echo "$@" > {args_file}\n{emit(bindir, CLI_RESULT)}')
    await speed.run_cli(str(path))
    args = args_file.read_text().split()
    assert "--accept-license" in args and "--accept-gdpr" in args and "--format=json" in args


async def test_run_cli_reports_why_it_failed(bindir):
    log = {"type": "log", "level": "error", "message": "Configuration - Couldn't resolve host name"}
    path = fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", f"{emit(bindir, log)}\nexit 1")
    with pytest.raises(SpeedError, match="Couldn't resolve host name"):
        await speed.run_cli(str(path))


async def test_run_cli_gives_up_after_the_timeout_and_kills_the_process(bindir, tmp_path):
    pid_file = tmp_path / "pid"
    path = fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", f"echo $$ > {pid_file}\nexec {SLEEP} 30")
    started = time.monotonic()
    with pytest.raises(SpeedError, match="took too long"):
        await speed.run_cli(str(path), timeout=0.5)
    assert time.monotonic() - started < 5
    pid = int(pid_file.read_text())
    await asyncio.sleep(0.2)
    with pytest.raises(ProcessLookupError):
        os.kill(pid, 0)


async def test_a_binary_that_cannot_start_is_a_speed_error(tmp_path):
    broken = tmp_path / "broken"
    broken.write_text("not executable")
    with pytest.raises(SpeedError):
        await speed.run_cli(str(broken))


# ---------------------------------------------------------------------------
# run_speedtest: CLI first, Python client second
# ---------------------------------------------------------------------------


async def test_the_cli_is_used_when_it_works(bindir, monkeypatch):
    fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", emit(bindir, CLI_RESULT))

    def boom():
        raise AssertionError("the Python client is not needed")

    monkeypatch.setattr(speed, "_run_python", boom)
    r = await speed.run_speedtest()
    assert r.source == "Ookla Speedtest CLI"


async def test_the_python_client_takes_over_when_the_cli_is_missing(bindir, monkeypatch):
    monkeypatch.setattr(speed, "_run_python", lambda: PY_RESULT)
    r = await speed.run_speedtest()
    assert "speedtest-cli" in r.source and round(r.download_mbps) == 245


async def test_the_python_client_takes_over_when_the_cli_fails(bindir, monkeypatch):
    fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", "exit 1")
    monkeypatch.setattr(speed, "_run_python", lambda: PY_RESULT)
    assert "speedtest-cli" in (await speed.run_speedtest()).source


async def test_when_both_fail_the_reasons_are_all_given(bindir, monkeypatch):
    fake_binary(bindir, "ookla-speedtest", "Speedtest by Ookla 1.2.0.84", "echo 'server unreachable'\nexit 1")

    def failing():
        raise SpeedError("speedtest-cli failed: HTTP Error 403: Forbidden")

    monkeypatch.setattr(speed, "_run_python", failing)
    with pytest.raises(SpeedError) as exc:
        await speed.run_speedtest()
    assert "server unreachable" in str(exc.value) and "403" in str(exc.value)


async def test_a_missing_python_client_is_explained(bindir, monkeypatch):
    monkeypatch.setitem(sys.modules, "speedtest", None)  # makes `import speedtest` fail, installed or not
    with pytest.raises(SpeedError) as exc:
        await speed.run_speedtest()
    assert "Speedtest CLI is not installed" in str(exc.value) and "speedtest-cli package is not installed" in str(exc.value)


async def test_errors_from_the_python_client_are_wrapped(bindir, monkeypatch):
    class FakeSpeedtest:
        def __init__(self, *a, **kw):
            raise RuntimeError("HTTP Error 403: Forbidden")

    monkeypatch.setitem(sys.modules, "speedtest", type(sys)("speedtest"))
    sys.modules["speedtest"].Speedtest = FakeSpeedtest
    with pytest.raises(SpeedError, match="speedtest-cli failed: HTTP Error 403"):
        await speed.run_speedtest()


async def test_a_python_client_that_hangs_is_given_up_on(bindir, monkeypatch):
    monkeypatch.setattr(speed, "CLI_TIMEOUT", 0.1)
    monkeypatch.setattr(speed, "_run_python", lambda: time.sleep(0.4) or PY_RESULT)
    with pytest.raises(SpeedError, match="took too long"):
        await speed.run_speedtest()


# ---------------------------------------------------------------------------
# the message
# ---------------------------------------------------------------------------


def test_format_result_lists_every_detail():
    text = speed.format_result(speed.parse_cli(CLI_RESULT))
    assert "SPEEDTEST" in text and "Ookla" in text
    assert "<code>250.00 Mbps</code>" in text and "<code>100.00 Mbps</code>" in text
    assert "<code>4.21 ms</code>" in text and "jitter <code>0.52 ms</code>" in text and "<code>0%</code>" in text
    assert "Hetzner Online GmbH" in text and "Example Telecom · Frankfurt, Germany" in text
    assert '<a href="https://www.speedtest.net/result/c/abc">speedtest.net</a>' in text


def test_format_result_skips_what_is_unknown_and_escapes_the_rest():
    r = SpeedResult(download_mbps=1234.5, upload_mbps=2.0, ping_ms=30, isp="A&B <ISP>", server="", location="", url="javascript:alert(1)")
    text = speed.format_result(r)
    assert "<code>1,234.50 Mbps</code>" in text and "A&amp;B &lt;ISP&gt;" in text
    assert "Server" not in text and "Packet loss" not in text and "jitter" not in text and "<a " not in text
    quoted = SpeedResult(download_mbps=1, upload_mbps=1, ping_ms=1, url='https://x.test/a"b')
    assert 'href="https://x.test/a&quot;b"' in speed.format_result(quoted)