# (c) cavaliba.com - tests / telemetry - get_keys, get_key_detail, get_value, get_series, get_raw

from datetime import UTC, datetime, timedelta

from django.test import TestCase

import app_data.telemetry as T
import app_home.cache as cache
from app_data.models import DataTelemetryAggregate
from tests.helper import TELEMETRY_T0, add_telemetry_raw

AT = TELEMETRY_T0 + timedelta(seconds=10)


class TelemetryQueryTest(TestCase):
    def setUp(self):
        cache.clear()
        T.reset_cache()
        add_telemetry_raw(TELEMETRY_T0, "auth.login", id="oidc")
        add_telemetry_raw(TELEMETRY_T0, "auth.login", id="local")
        add_telemetry_raw(TELEMETRY_T0, "x.t", "timer", 40)
        for value in [5, 50]:
            add_telemetry_raw(TELEMETRY_T0, "x.h", "histogram", value)
        T.telemetry_aggregate(now=TELEMETRY_T0 + timedelta(minutes=3))

    def test_get_keys(self):
        keys = {k["key"]: k for k in T.get_keys()}
        self.assertTrue(set(T.TELEMETRY_KEYS) <= set(keys))
        self.assertEqual((keys["x.t"]["type"], keys["x.t"]["declared"]), ("timer", False))
        self.assertTrue(keys["auth.login"]["declared"])

    def test_get_key_detail(self):
        detail = T.get_key_detail("auth.login")
        self.assertEqual(detail["functions"], ["sum", "count"])
        self.assertEqual(detail["default_function"], "sum")
        self.assertEqual(detail["ids"], ["local", "oidc"])
        self.assertEqual(detail["ranges"]["minute"]["count"], 1)
        self.assertEqual(T.get_key_detail("users.active")["periods"], ["day", "week", "month"])

    def test_get_value(self):
        self.assertEqual(T.get_value("auth.login", "sum", "minute", at=AT)["value"], 2)
        self.assertEqual(T.get_value("auth.login", None, "minute", at=AT, id="oidc")["value"], 1)
        self.assertEqual(T.get_value("x.t", None, "minute", at=AT)["value"], 40)

    def test_missing_bucket_values(self):
        before = TELEMETRY_T0 - timedelta(hours=1)
        self.assertEqual(T.get_value("auth.login", "sum", "minute", at=before)["value"], 0)
        self.assertEqual(T.get_value("x.t", "count", "minute", at=before)["value"], 0)
        self.assertIsNone(T.get_value("x.t", "p95", "minute", at=before)["value"])
        self.assertEqual(T.get_value("x.h", "buckets", "minute", at=before)["value"], {})

    def test_histogram_buckets_in_bound_order(self):
        value = T.get_value("x.h", None, "minute", at=AT)["value"]
        self.assertEqual(list(value), ["le_1", "le_10", "le_100", "le_1000", "le_10000", "inf"])

    def test_get_series_fills_gaps(self):
        series = T.get_series(
            "auth.login",
            "sum",
            "minute",
            start=AT - timedelta(minutes=2),
            end=AT + timedelta(minutes=1),
        )
        self.assertEqual([p["value"] for p in series], [0, 0, 2, 0])
        self.assertEqual(series[2]["start"], TELEMETRY_T0)

    def test_series_dst_fall_back_hours(self):
        # 2026-10-25: 02:00 local occurs twice (00:00 UTC CEST, 01:00 UTC CET)
        for hour, count in [(0, 3), (1, 5)]:
            DataTelemetryAggregate.objects.create(
                key="auth.login", period="hour", count=count, sum=count,
                start=datetime(2026, 10, 25, hour, 0, tzinfo=UTC),
            )  # fmt: skip
        series = T.get_series(
            "auth.login", "sum", "hour",
            start="2026-10-25T00:00+00:00", end="2026-10-25T02:00+00:00",
        )  # fmt: skip
        self.assertEqual([p["value"] for p in series], [3, 5, 0])

    def test_query_errors(self):
        with self.assertRaises(T.TelemetryNotFound):
            T.get_value("no.key", "sum", "minute")
        with self.assertRaises(T.TelemetryError):
            T.get_value("auth.login", "p95", "minute")
        with self.assertRaises(T.TelemetryError):
            T.get_value("users.active", "distinct", "hour")
        with self.assertRaises(T.TelemetryError):
            T.get_series("auth.login", "sum", "minute", start="now", end="now-1h")
        with self.assertRaises(T.TelemetryError):
            T.get_series("auth.login", "sum", "minute", start="now-1w", end="now")

    def test_get_raw(self):
        raws = T.get_raw(limit=2, key="auth.login")
        self.assertEqual(len(raws), 2)
        self.assertEqual({r["key"] for r in raws}, {"auth.login"})