mirror of
https://github.com/Horizontal42/netsleuth.git
synced 2026-11-10 23:52:04 +03:00
116 lines
3.1 KiB
Python
116 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
from pytest_httpx import HTTPXMock
|
|
|
|
from netsleuth.models import AnycastHop, PathDiversity
|
|
from netsleuth.probes.bgp_path import (
|
|
build_path_diversity,
|
|
colo_location,
|
|
detect_international_loop,
|
|
parse_cf_ray,
|
|
probe_edge,
|
|
)
|
|
|
|
_CYRILLIC = set("абвгдеёжзийклмнопрстуфхцчшщъыьэюя")
|
|
|
|
|
|
def has_cyrillic(text: str) -> bool:
|
|
return any(ch in _CYRILLIC for ch in (text or "").lower())
|
|
|
|
|
|
def test_parse_cf_ray_extracts_colo_code():
|
|
assert parse_cf_ray("9a1b2c3d4e5f6789-DME") == "DME"
|
|
|
|
|
|
def test_parse_cf_ray_normalizes_lowercase():
|
|
assert parse_cf_ray("9a1b2c3d4e5f6789-dme") == "DME"
|
|
|
|
|
|
def test_parse_cf_ray_none_input():
|
|
assert parse_cf_ray(None) is None
|
|
|
|
|
|
def test_parse_cf_ray_empty_string():
|
|
assert parse_cf_ray("") is None
|
|
|
|
|
|
def test_parse_cf_ray_no_dash():
|
|
assert parse_cf_ray("nodash") is None
|
|
|
|
|
|
def test_parse_cf_ray_trailing_dash_empty_suffix():
|
|
assert parse_cf_ray("trailing-") is None
|
|
|
|
|
|
def test_colo_location_known_code():
|
|
city, country = colo_location("DME")
|
|
assert country == "RU"
|
|
assert city
|
|
|
|
|
|
def test_colo_location_unknown_code():
|
|
assert colo_location("ZZZ") == (None, None)
|
|
|
|
|
|
def test_colo_location_none():
|
|
assert colo_location(None) == (None, None)
|
|
|
|
|
|
def test_detect_international_loop_true_when_edge_diverges():
|
|
assert detect_international_loop("RU", "DE") == (True, ["DE"])
|
|
|
|
|
|
def test_detect_international_loop_false_when_edge_matches_client():
|
|
assert detect_international_loop("RU", "RU") == (False, [])
|
|
|
|
|
|
def test_detect_international_loop_false_when_edge_unknown():
|
|
assert detect_international_loop("RU", None) == (False, [])
|
|
|
|
|
|
def test_detect_international_loop_false_when_client_country_missing():
|
|
assert detect_international_loop(None, "DE") == (False, [])
|
|
|
|
|
|
def test_build_path_diversity_with_looping_hop():
|
|
hop = AnycastHop(target="example.com", edge_country="DE", source="cf_ray")
|
|
diversity = build_path_diversity("RU", [hop])
|
|
assert diversity.international_loop is True
|
|
assert diversity.detour_countries
|
|
assert has_cyrillic(diversity.note_ru)
|
|
|
|
|
|
def test_build_path_diversity_with_empty_hops_is_default():
|
|
assert build_path_diversity("RU", []) == PathDiversity(client_country="RU", hops=[])
|
|
|
|
|
|
def test_anycast_hop_rejects_unknown_source():
|
|
with pytest.raises(ValueError):
|
|
AnycastHop(target="x", source="bogus")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_edge_uses_cf_ray_header(httpx_mock: HTTPXMock):
|
|
httpx_mock.add_response(
|
|
url="https://example.com/cdn-cgi/trace",
|
|
headers={"cf-ray": "abc123-FRA"},
|
|
text="ip=1.2.3.4\ncolo=FRA\n",
|
|
)
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
hop = await probe_edge(client, "example.com", timeout=2.0)
|
|
|
|
assert hop.edge_colo == "FRA"
|
|
assert hop.source == "cf_ray"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_probe_edge_swallows_connection_error(httpx_mock: HTTPXMock):
|
|
httpx_mock.add_exception(httpx.ConnectError("boom"))
|
|
|
|
async with httpx.AsyncClient() as client:
|
|
hop = await probe_edge(client, "example.com", timeout=2.0)
|
|
|
|
assert hop == AnycastHop(target="example.com", source="none")
|