"""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