# Copyright © The Debusine Developers
# See the AUTHORS file at the top-level directory of this distribution
#
# This file is part of Debusine. It is subject to the license terms
# in the LICENSE file found in the top-level directory of this
# distribution. No part of Debusine, including this file, may be copied,
# modified, propagated, or distributed except according to the terms
# contained in the LICENSE file.

"""Unit tests for Redis Prometheus Values."""

import datetime as dt
import time as tm
from typing import cast

from django.test import SimpleTestCase
from prometheus_client.samples import Exemplar

from debusine.server.prometheus_client.redis import (
    _daemon_threads,
    _live_metrics,
    private_fake_redis,
    redis_client,
)
from debusine.server.prometheus_client.values import (
    MULTIPROCESS_MODE_T,
    RedisValue,
    Value,
)


class RedisValueProtocol(Value):
    """Extend the Value protocol to give us access to RedisValue internals."""

    _typ: str
    _metric_name: str
    _name: str
    _labelnames: list[str]
    _labelvalues: list[str]
    _help_text: str
    _multiprocess_mode: MULTIPROCESS_MODE_T
    _expiry: dt.timedelta | None
    _key: str


@private_fake_redis()
class ValueTestCase(SimpleTestCase):
    """Test Redis values."""

    pid: int = 123

    def create_value(
        self,
        metric_name: str = "test",
        name: str | None = None,
        type_: str = "counter",
        labelnames: list[str] | None = None,
        labelvalues: list[str] | None = None,
        multiprocess_mode: MULTIPROCESS_MODE_T = "",
    ) -> RedisValueProtocol:
        value = RedisValue(lambda: self.pid)(
            type_,
            metric_name,
            name or metric_name + "_total",
            labelnames or [],
            labelvalues or [],
            "Help Text",
            multiprocess_mode=multiprocess_mode,
        )
        return cast(RedisValueProtocol, value)

    def test_initializes_value(self) -> None:
        value = self.create_value()
        self.assertEqual(value.get(), 0.0)

    def test_sets_and_gets_value(self) -> None:
        value = self.create_value()
        value.set(5)
        self.assertEqual(value.get(), 5.0)

    def test_inc_value(self) -> None:
        value = self.create_value()
        value.inc(3)
        value.inc(5)
        self.assertEqual(value.get(), 8.0)

    def test_get_missing_value(self) -> None:
        value = self.create_value()
        redis_client().delete(value._key)
        self.assertEqual(value.get(), 0.0)

    def test_exemplars_not_implemented(self) -> None:
        value = self.create_value("test4")
        with self.assertRaises(NotImplementedError):
            value.set_exemplar(Exemplar(labels={}, value=0.0, timestamp=0.0))
        self.assertIsNone(value.get_exemplar())

    def test_differentiated_by_name(self) -> None:
        v1 = self.create_value("value1")
        v2 = self.create_value("value2")
        v1.set(1)
        v2.set(2)
        self.assertEqual(v1.get(), 1.0)
        self.assertEqual(v2.get(), 2.0)

    def test_differentiated_by_labels(self) -> None:
        v1 = self.create_value("value3", labelnames=["a"], labelvalues=["1"])
        v2 = self.create_value("value3", labelnames=["a"], labelvalues=["2"])
        v1.set(1)
        v2.set(2)
        self.assertEqual(v1.get(), 1.0)
        self.assertEqual(v2.get(), 2.0)

    def test_multiprocess_mode_mostrecent(self) -> None:
        with self.assertRaises(NotImplementedError):
            self.create_value(type_="gauge", multiprocess_mode="mostrecent")

    def test_multiprocess_mode_counter(self) -> None:
        with self.assertRaises(AssertionError):
            self.create_value(type_="counter", multiprocess_mode="liveall")

    def test_multiprocess_mode(self) -> None:
        value = self.create_value(type_="gauge", multiprocess_mode="all")
        self.assertEqual(value._labelnames, ["pid"])
        self.assertEqual(value._labelvalues[-1], "123")
        self.assertIsNone(value._expiry)
        self.assertLess(redis_client().expiretime(value._key), 0)

    def test_multiprocess_mode_live(self) -> None:
        value = self.create_value(type_="gauge", multiprocess_mode="liveall")
        unixtime = tm.time()
        self.assertEqual(value._labelnames, ["pid"])
        self.assertEqual(value._labelvalues[-1], "123")
        self.assertEqual(value._expiry, dt.timedelta(seconds=20))
        expiretime = redis_client().expiretime(value._key)
        self.assertGreater(expiretime, unixtime)
        self.assertLessEqual(expiretime, unixtime + 20)

        self.assertIn("123", _daemon_threads)
        self.assertIn("123", _live_metrics)
        self.assertIn(value._key, _live_metrics["123"])

    def test_live_inc_updates_expiry(self) -> None:
        value = self.create_value(type_="gauge", multiprocess_mode="liveall")
        unixtime = tm.time()
        redis_client().persist(value._key)
        self.assertLess(redis_client().expiretime(value._key), 0)

        value.inc(1)
        self.assertGreater(redis_client().expiretime(value._key), unixtime)

    def test_live_set_updates_expiry(self) -> None:
        value = self.create_value(type_="gauge", multiprocess_mode="liveall")
        unixtime = tm.time()
        redis_client().persist(value._key)
        self.assertLess(redis_client().expiretime(value._key), 0)

        value.set(1)
        self.assertGreater(redis_client().expiretime(value._key), unixtime)

    def test_live_shares_daemon_thread(self) -> None:
        self.create_value("value1", type_="gauge", multiprocess_mode="liveall")
        self.create_value("value2", type_="gauge", multiprocess_mode="liveall")

    def test_multiprocess_pid_change(self) -> None:
        self.pid = 1

        value = self.create_value(type_="gauge", multiprocess_mode="all")
        self.assertEqual(value._labelnames[-1], "pid")
        self.assertEqual(value._labelvalues[-1], "1")
        value.inc(1)
        self.assertEqual(value.get(), 1.0)

        self.pid = 2
        value.inc(1)
        self.assertEqual(value._labelvalues[-1], "2")
        self.assertEqual(value.get(), 1.0)
