mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-11 01:02:10 +03:00
84c3dffbce
New Phase 0 in diagnose(): three generate_204-style probes run before every other module, using detect-and-tag rather than short-circuit — a portal is rarely total (DNS/ICMP often pass while only HTTP is intercepted), and an aborted run would look identical to a dead link. On detection every later phase still runs, tagged with a crit finding and a banner at the top of the report. Caught live before it shipped: detectportal.firefox.com/success.txt (one of the original three default check URLs) returns 200 "success" on a genuinely clean connection, which the generate_204-shaped classifier misread as a portal redirect. Swapped for a second real generate_204 endpoint instead of teaching the classifier a second success shape. Also redesigned the speedtest server geo-sanity check after checking the bundled Ookla CLI's actual JSON output: its server object never carries lat/lon (documented schema, not a version gap), so speed.server_other_country compares server_country against the client's own country instead of a haversine distance that no tier can actually supply. Part of plans/zesty-zooming-wilkinson.md (gitignored).
129 lines
3.8 KiB
Python
129 lines
3.8 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from netsleuth.config import Speedtest
|
|
from netsleuth.speed import tier_fastcom
|
|
|
|
|
|
async def test_tier_fastcom_happy_path(httpx_mock):
|
|
config = Speedtest()
|
|
api_url = config.fastcom_api_url
|
|
|
|
httpx_mock.add_response(
|
|
url=f"{api_url}?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3",
|
|
json={
|
|
"targets": [
|
|
{"url": "https://server1.example.com/speedtest", "location": {"country": "US"}},
|
|
{"url": "https://server2.example.com/speedtest", "location": {"country": "US"}},
|
|
]
|
|
},
|
|
)
|
|
|
|
httpx_mock.add_response(
|
|
url="https://server1.example.com/speedtest",
|
|
content=b"A" * 125000, # 1 Mbit
|
|
)
|
|
|
|
httpx_mock.add_response(
|
|
url="https://server2.example.com/speedtest",
|
|
content=b"B" * 250000, # 2 Mbit
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await tier_fastcom(client, config, timeout=5.0)
|
|
|
|
assert result.method == "fastcom"
|
|
assert result.download_mbps is not None
|
|
assert result.download_mbps > 0
|
|
assert result.upload_mbps is None
|
|
assert result.server == "server1.example.com"
|
|
assert result.netflix_oca_onnet is True
|
|
assert result.server_country == "US"
|
|
|
|
|
|
async def test_tier_fastcom_no_targets(httpx_mock):
|
|
config = Speedtest()
|
|
api_url = config.fastcom_api_url
|
|
|
|
httpx_mock.add_response(
|
|
url=f"{api_url}?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3",
|
|
json={"targets": []},
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await tier_fastcom(client, config, timeout=5.0)
|
|
|
|
assert result.method == "fastcom"
|
|
assert result.download_mbps == 0.0
|
|
assert result.server is None
|
|
assert result.netflix_oca_onnet is None
|
|
|
|
|
|
async def test_tier_fastcom_missing_urls(httpx_mock):
|
|
config = Speedtest()
|
|
api_url = config.fastcom_api_url
|
|
|
|
httpx_mock.add_response(
|
|
url=f"{api_url}?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3",
|
|
json={
|
|
"targets": [
|
|
{"location": {"country": "US"}}, # missing URL
|
|
{"url": "https://server1.example.com/speedtest", "location": {"country": "US"}},
|
|
]
|
|
},
|
|
)
|
|
|
|
httpx_mock.add_response(
|
|
url="https://server1.example.com/speedtest",
|
|
content=b"A" * 125000, # 1 Mbit
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
result = await tier_fastcom(client, config, timeout=5.0)
|
|
|
|
assert result.method == "fastcom"
|
|
assert result.download_mbps is not None
|
|
assert result.download_mbps > 0
|
|
assert result.server is None
|
|
assert result.netflix_oca_onnet is True
|
|
|
|
|
|
async def test_tier_fastcom_api_error(httpx_mock):
|
|
config = Speedtest()
|
|
api_url = config.fastcom_api_url
|
|
|
|
httpx_mock.add_response(
|
|
url=f"{api_url}?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3",
|
|
status_code=500,
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
with pytest.raises(httpx.HTTPStatusError) as exc:
|
|
await tier_fastcom(client, config, timeout=5.0)
|
|
|
|
assert exc.value.response.status_code == 500
|
|
|
|
|
|
async def test_tier_fastcom_download_error(httpx_mock):
|
|
config = Speedtest()
|
|
api_url = config.fastcom_api_url
|
|
|
|
httpx_mock.add_response(
|
|
url=f"{api_url}?https=true&token=YXNkZmFzZGxmbnNkYWZoYXNkZmhrYWxm&urlCount=3",
|
|
json={
|
|
"targets": [
|
|
{"url": "https://server1.example.com/speedtest", "location": {"country": "US"}},
|
|
]
|
|
},
|
|
)
|
|
|
|
httpx_mock.add_response(
|
|
url="https://server1.example.com/speedtest",
|
|
status_code=404,
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
with pytest.raises(httpx.HTTPStatusError) as exc:
|
|
await tier_fastcom(client, config, timeout=5.0)
|
|
|
|
assert exc.value.response.status_code == 404
|