Spaces:
Running
Running
File size: 3,272 Bytes
5f400f8 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | """/speed: the command around the speed test."""
import asyncio
from bot import app as appmod
from bot import speed
from bot.speed import SpeedResult
from .appkit import ADMIN_CFG, OWNER, STRANGER, say
RESULT = SpeedResult(
download_mbps=250.0, upload_mbps=100.0, ping_ms=4.2, jitter_ms=0.5, packet_loss=0.0, isp="Hetzner",
server="Example Telecom", location="Frankfurt, Germany", url="https://www.speedtest.net/result/c/abc", source="Ookla Speedtest CLI",
)
def serve(monkeypatch, result=RESULT, error=None, gate=None):
calls = []
async def fake():
calls.append(1)
if gate is not None:
await gate.wait()
if error is not None:
raise error
return result
monkeypatch.setattr(speed, "run_speedtest", fake)
return calls
async def test_speed_shows_a_status_and_then_the_figures(ctx, monkeypatch):
calls = serve(monkeypatch)
ev = await say(ctx, OWNER, "/speed")
assert ev.texts == ["⚡ Running the speed test… this takes about half a minute."]
(status, text, kw), = ctx.client.edited
assert status.id == 1001 and kw["parse_mode"] == "html" and kw["link_preview"] is False
assert "SPEEDTEST" in text and "250.00 Mbps" in text and "100.00 Mbps" in text and "speedtest.net/result/c/abc" in text
assert len(calls) == 1 and "may be lower" not in text
async def test_a_failed_test_says_why(ctx, monkeypatch):
serve(monkeypatch, error=speed.SpeedError("the Speedtest CLI is not installed; the speedtest-cli package is not installed <x>"))
await say(ctx, OWNER, "/speed")
text = ctx.client.edited[-1][1]
assert text.startswith("❌ I couldn't run the speed test: the Speedtest CLI is not installed") and "<x>" in text
async def test_an_unexpected_crash_is_reported_and_does_not_leave_the_lock_taken(ctx, monkeypatch):
serve(monkeypatch, error=RuntimeError("boom"))
await say(ctx, OWNER, "/speed")
assert "failed unexpectedly" in ctx.client.edited[-1][1]
serve(monkeypatch)
await say(ctx, OWNER, "/speed")
assert "250.00 Mbps" in ctx.client.edited[-1][1]
async def test_only_one_test_runs_at_a_time(ctx, monkeypatch):
gate = asyncio.Event()
calls = serve(monkeypatch, gate=gate)
first = asyncio.ensure_future(say(ctx, OWNER, "/speed"))
await asyncio.sleep(0.05)
second = await say(ctx, ADMIN_CFG, "/speed")
assert "already running" in second.texts[0] and len(calls) == 1
gate.set()
await first
assert "250.00 Mbps" in ctx.client.edited[-1][1]
again = await say(ctx, ADMIN_CFG, "/speed") # free again
assert again.texts[0].startswith("⚡")
async def test_the_result_mentions_a_running_job(ctx, monkeypatch):
serve(monkeypatch)
await say(ctx, OWNER, "/speed")
assert "A rename job was running" not in ctx.client.edited[-1][1]
monkeypatch.setattr(type(ctx.app.queue), "running", property(lambda self: 1))
await say(ctx, OWNER, "/speed")
assert "A rename job was running during the test" in ctx.client.edited[-1][1]
async def test_strangers_cannot_run_it_and_it_is_in_the_menu(ctx, monkeypatch):
calls = serve(monkeypatch)
assert (await say(ctx, STRANGER, "/speed")).out == [] and calls == []
assert "speed" in dict(appmod.COMMANDS)
|