# (c) cavaliba.com - tests / telemetry - catch-up windows, purge watermarks, gauge,
# quarter/year rollup, id folding at rollup level

from datetime import timedelta

from django.test import TestCase

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


def _agg(key, period, did=""):
    return DataTelemetryAggregate.objects.get(key=key, period=period, distinct_id=did)


class TelemetryCatchupTest(TestCase):
    def setUp(self):
        cache.clear()
        T.reset_cache()

    def _run_until(self, end, step=timedelta(minutes=30)):
        now = TELEMETRY_T0
        while now < end:
            now += step
            T.telemetry_aggregate(now=now)

    def test_long_window_chunked_into_minute_buckets(self):
        # 18 events over 3 hours: one pass covers them all (240 minute buckets), split in
        # windows of RAW_CHUNK_BUCKETS; every minute bucket must still be written once
        for i in range(18):
            add_telemetry_raw(TELEMETRY_T0 + timedelta(minutes=10 * i), "x.c", value=2)

        T.telemetry_aggregate(now=TELEMETRY_T0 + timedelta(hours=3, minutes=5))

        minutes = DataTelemetryAggregate.objects.filter(key="x.c", period="minute")
        self.assertEqual(minutes.count(), 18)
        self.assertEqual({m.sum for m in minutes}, {2})
        hours = DataTelemetryAggregate.objects.filter(key="x.c", period="hour").order_by("start")
        self.assertEqual([h.sum for h in hours], [12, 12, 12])
        # watermark = first bucket not finalized: the one holding now - GRACE_SECONDS
        closed = TELEMETRY_T0 + timedelta(hours=3, minutes=5) - timedelta(seconds=T.GRACE_SECONDS)
        self.assertEqual(T._wm_get("minute"), T.bucket_start(closed, "minute"))

    def test_purge_waits_for_the_lowest_raw_watermark(self):
        # hour aggregation ran ahead (catch-up), minute aggregation did not: the raw row
        # must survive until the minute watermark passes it
        add_telemetry_raw(TELEMETRY_T0, "x.c")
        T._wm_set("minute", TELEMETRY_T0)
        T._wm_set("hour", TELEMETRY_T0 + timedelta(hours=2))

        T._purge_raw(TELEMETRY_T0 + timedelta(hours=2))

        self.assertEqual(DataTelemetryRaw.objects.count(), 1)

        T._wm_set("minute", TELEMETRY_T0 + timedelta(hours=2))
        T._purge_raw(TELEMETRY_T0 + timedelta(hours=2))

        self.assertEqual(DataTelemetryRaw.objects.count(), 0)

    def test_purge_hard_cap_still_applies(self):
        add_telemetry_raw(TELEMETRY_T0, "x.c")
        T._wm_set("minute", TELEMETRY_T0)
        T._wm_set("hour", TELEMETRY_T0)

        T._purge_raw(TELEMETRY_T0 + timedelta(hours=T.RAW_HARD_KEEP_HOURS + 1))

        self.assertEqual(DataTelemetryRaw.objects.count(), 0)

    def test_gauge_hour_and_day(self):
        for i, value in enumerate([5, 9, 7]):
            add_telemetry_raw(TELEMETRY_T0 + timedelta(minutes=i), "x.g", "gauge", value)

        self._run_until(TELEMETRY_T0 + timedelta(hours=3))

        hour = _agg("x.g", "hour")
        self.assertEqual((hour.count, hour.last, hour.min, hour.max, hour.mean), (3, 7, 5, 9, 7))
        day = _agg("x.g", "day")
        self.assertEqual((day.count, day.last, day.min, day.max, day.mean), (3, 7, 5, 9, 7))

    def test_quarter_and_year_rollup(self):
        add_telemetry_raw(TELEMETRY_T0, "x.c", value=4)
        self._run_until(TELEMETRY_T0 + timedelta(hours=3))

        # 10-day steps: each pass moves the hour watermark forward, then day, month, quarter
        self._run_until(TELEMETRY_T0 + timedelta(days=90), step=timedelta(days=9))

        self.assertEqual(_agg("x.c", "quarter").sum, 4)
        self.assertEqual(_agg("x.c", "year").sum, 4)
        self.assertEqual(_agg("x.c", "quarter").start, T.bucket_start(TELEMETRY_T0, "quarter"))

    def test_rollup_folds_ids_into_other(self):
        for i in range(T.DEFAULT_MAX_IDS + 5):
            for _ in range(i + 1):
                add_telemetry_raw(TELEMETRY_T0, "x.c", id=f"id{i:02d}")

        self._run_until(TELEMETRY_T0 + timedelta(hours=3))

        total = sum(range(1, T.DEFAULT_MAX_IDS + 6))
        for period in ("hour", "day"):
            ids = DataTelemetryAggregate.objects.filter(key="x.c", period=period)
            ids = ids.exclude(distinct_id="")
            # at most max_ids kept plus _other; rolling up an already folded hour may fold
            # the lightest id into the existing _other, leaving one row less
            self.assertLessEqual(ids.count(), T.DEFAULT_MAX_IDS + 1, period)
            self.assertGreaterEqual(ids.count(), T.DEFAULT_MAX_IDS, period)
            self.assertIn("_other", [row.distinct_id for row in ids])
            # no event lost by folding: ids sum to the total row
            self.assertEqual(sum(row.sum for row in ids), total, period)
            self.assertEqual(_agg("x.c", period).sum, total, period)