thermograph/tests/test_metrics.py

"""Unit tests for the in-process traffic counters (metrics.py)."""
import importlib

import metrics


def _fresh():
    # Counters are module-level; reload for an isolated starting state per test.
    return importlib.reload(metrics)


def test_phase_source_map_covers_every_source():
    m = _fresh()
    assert m.source_for_phase("history_fetch") == "open-meteo-archive"
    assert m.source_for_phase("history_topup") == "open-meteo-archive"
    assert m.source_for_phase("recent_forecast_fetch") == "open-meteo-forecast"
    assert m.source_for_phase("geocode") == "open-meteo-geocoding"
    assert m.source_for_phase("history_nasa") == "nasa-power"
    assert m.source_for_phase("forecast_metno") == "met-norway"
    assert m.source_for_phase("reverse_geocode") == "nominatim"
    assert m.source_for_phase("unknown-phase") == "other"


def test_classify_inbound_categories():
    m = _fresh()
    c = m.classify_inbound
    # BASE stripped, API namespaces
    assert c("/thermograph/api/v2/place", "/thermograph") == "api:place"
    assert c("/thermograph/api/v2/grade", "/thermograph") == "api:grade"
    assert c("/thermograph/api/v1/geocode", "/thermograph") == "api:geocode"
    assert c("/thermograph/api/v2/auth/login", "/thermograph") == "auth"
    assert c("/thermograph/api/v2/users/me", "/thermograph") == "auth"
    assert c("/thermograph/api/v2/subscriptions", "/thermograph") == "accounts"
    assert c("/thermograph/api/v2/notifications", "/thermograph") == "accounts"
    assert c("/thermograph/api/v2/push/subscribe", "/thermograph") == "accounts"
    assert c("/thermograph/api/v2/metrics", "/thermograph") == "metrics"
    # pages, static, seo
    assert c("/thermograph/", "/thermograph") == "page"
    assert c("/thermograph/calendar", "/thermograph") == "page"
    assert c("/thermograph/climate/tokyo", "/thermograph") == "page"
    assert c("/thermograph/app.js", "/thermograph") == "static"
    assert c("/thermograph/style.css", "/thermograph") == "static"
    assert c("/thermograph/robots.txt", "/thermograph") == "seo"
    # root-mounted (BASE == "")
    assert c("/api/v2/place", "") == "api:place"
    assert c("/", "") == "page"


def test_record_inbound_buckets_by_status():
    m = _fresh()
    m.record_inbound("api:place", 200)
    m.record_inbound("api:place", 200)
    m.record_inbound("api:place", 503)
    m.record_inbound("page", 304)
    snap = m.snapshot()
    assert snap["inbound"]["api:place"]["total"] == 3
    assert snap["inbound"]["api:place"]["2xx"] == 2
    assert snap["inbound"]["api:place"]["5xx"] == 1
    assert snap["inbound"]["page"]["3xx"] == 1
    assert snap["inbound_total"] == 4


def test_metrics_endpoint_polling_is_not_counted():
    # The dashboard polls the metrics endpoint; that self-traffic must be excluded.
    m = _fresh()
    m.record_inbound("metrics", 200)
    m.record_inbound("metrics", 404)
    m.record_inbound("api:place", 200)
    snap = m.snapshot()
    assert "metrics" not in snap["inbound"]
    assert snap["inbound_total"] == 1


def test_record_outbound_by_source_and_outcome():
    m = _fresh()
    m.record_outbound("history_fetch", "ok")
    m.record_outbound("history_topup", "ok")          # same source (archive)
    m.record_outbound("reverse_geocode", "error")
    m.record_outbound("recent_forecast_fetch", "rate_limited")
    snap = m.snapshot()
    assert snap["outbound"]["open-meteo-archive"]["ok"] == 2
    assert snap["outbound"]["nominatim"]["error"] == 1
    assert snap["outbound"]["open-meteo-forecast"]["rate_limited"] == 1
    assert snap["outbound_total"] == 4


def test_snapshot_has_process_facts():
    m = _fresh()
    snap = m.snapshot()
    assert isinstance(snap["pid"], int)
    assert snap["uptime_s"] >= 0
    assert "open-meteo-archive" in snap["outbound"]  # stable source rows exist


def test_recent_window_sums_and_rolls_off(monkeypatch):
    m = _fresh()
    clock = {"now": 1_000_000.0}
    monkeypatch.setattr(m.time, "time", lambda: clock["now"])

    # minute 0: three page hits + one archive call
    for _ in range(3):
        m.record_inbound("page", 200)
    m.record_outbound("history_fetch", "ok")          # -> open-meteo-archive
    snap = m.snapshot()
    assert snap["window_minutes"] == 10
    assert snap["inbound_recent"]["page"] == 3
    assert snap["outbound_recent"]["open-meteo-archive"] == 1

    # +5 min: still inside the 10-minute window, so the earlier hits still count
    clock["now"] += 5 * 60
    m.record_inbound("page", 200)
    assert m.snapshot()["inbound_recent"]["page"] == 4

    # +6 more min (t = 11 min): the minute-0 burst has rolled out of the window;
    # only the hit from minute 5 remains, and the archive call is gone entirely.
    clock["now"] += 6 * 60
    snap = m.snapshot()
    assert snap["inbound_recent"].get("page", 0) == 1
    assert "open-meteo-archive" not in snap["outbound_recent"]
    # the cumulative since-start total is unaffected by the rolling window
    assert snap["inbound"]["page"]["total"] == 4


def test_recent_excludes_self_polling():
    m = _fresh()
    m.record_inbound("metrics", 200)                  # dashboard polling — never counted
    m.record_inbound("page", 200)
    snap = m.snapshot()
    assert "metrics" not in snap["inbound_recent"]
    assert snap["inbound_recent"]["page"] == 1