tests/unit/test_scoring_dns.py (view raw)
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 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 |
"""DNS fingerprint helpers and scoring / smart list."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from app.db.models import ProbeResult, Tracker
from app.services.dns import compute_fingerprint
from app.services.scoring import compute_score, latency_score, metrics_from_results
from app.services.smart_list import SmartListEntry, SmartListFilters, format_smart_list_txt
def test_cname_fingerprint_deterministic():
a = compute_fingerprint("cdn.example.net", ["1.2.3.4"])
b = compute_fingerprint("cdn.example.net", ["9.9.9.9"])
assert a == b
ips = compute_fingerprint(None, ["2.2.2.2", "1.1.1.1"])
ips2 = compute_fingerprint(None, ["1.1.1.1", "2.2.2.2"])
assert ips == ips2
assert a != ips
def test_scoring_and_provisional():
assert latency_score(100) == 20
assert latency_score(250) == 16
assert latency_score(2000) == 0
assert (
compute_score(uptime_7d=1.0, latency_median_7d=100, valid_rate_7d=1.0, measurement_count=2)
is None
)
score = compute_score(
uptime_7d=1.0, latency_median_7d=100, valid_rate_7d=1.0, measurement_count=25
)
assert score == 100
tracker = Tracker(
canonical_url="udp://t.example:80/announce",
scheme="udp",
hostname="t.example",
port=80,
path="/announce",
current_status="up",
first_seen_at=datetime.now(UTC) - timedelta(days=3),
last_seen_at=datetime.now(UTC),
consecutive_failures=0,
)
now = datetime.now(UTC)
results = [
ProbeResult(
tracker_id=1,
checked_at=now - timedelta(hours=i),
status="up",
latency_ms=120,
response_valid=True,
)
for i in range(5)
]
m = metrics_from_results(tracker, results, now=now)
assert m.provisional is True
assert m.score is not None
assert m.uptime_7d == 1.0
def test_smart_list_txt_blank_lines():
t1 = Tracker(
id=1,
canonical_url="udp://a.example:80/announce",
scheme="udp",
hostname="a.example",
port=80,
path="/announce",
current_status="up",
first_seen_at=datetime.now(UTC),
last_seen_at=datetime.now(UTC),
consecutive_failures=0,
)
t2 = Tracker(
id=2,
canonical_url="udp://b.example:80/announce",
scheme="udp",
hostname="b.example",
port=80,
path="/announce",
current_status="up",
first_seen_at=datetime.now(UTC),
last_seen_at=datetime.now(UTC),
consecutive_failures=0,
)
from app.services.scoring import TrackerMetrics
metrics = TrackerMetrics(
uptime_24h=1,
uptime_7d=1,
uptime_30d=1,
latency_median_7d=10,
latency_p95_7d=20,
valid_rate_7d=1,
measurement_count=20,
measurement_count_7d=20,
tracking_days=10,
score=90,
provisional=False,
)
txt = format_smart_list_txt([SmartListEntry(t1, metrics), SmartListEntry(t2, metrics)])
assert txt == "udp://a.example:80/announce\n\nudp://b.example:80/announce\n"
assert SmartListFilters().limit == 20
|