# 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.

"""Tests for the prometheus redis collector."""

from typing import override

from django.test import SimpleTestCase
from prometheus_client import values
from prometheus_client.core import CollectorRegistry, Sample
from prometheus_client.metrics import Counter, Gauge, Histogram

from debusine.server.prometheus_client.redis import (
    _daemon_threads,
    mark_process_dead,
    private_fake_redis,
    redis_client,
)
from debusine.server.prometheus_client.redis_collector import RedisCollector
from debusine.server.prometheus_client.values import RedisValue


@private_fake_redis()
class RedisCollectorTestCase(SimpleTestCase):
    """Test Redis collector."""

    @override
    def setUp(self) -> None:
        super().setUp()
        self.registry = CollectorRegistry()
        self.collector = RedisCollector(self.registry)
        redis_client().flushdb()

        old_values_class = values.ValueClass

        def restore_old_values_class() -> None:
            values.ValueClass = old_values_class

        values.ValueClass = RedisValue(lambda: 123)
        self.addCleanup(restore_old_values_class)

    @override
    def tearDown(self) -> None:
        super().tearDown()
        for identifier in list(_daemon_threads):
            mark_process_dead(identifier)

    def test_create_collector(self) -> None:
        collector = RedisCollector(None)
        self.assertEqual(list(collector.collect()), [])

    def test_counter(self) -> None:
        counter = Counter(
            "test",
            "Test Counter",
            labelnames=("label",),
            registry=self.registry,
        )
        counter.labels(label="a").inc()
        counter.labels(label="b").inc(7)

        metrics = {m.name: m for m in self.collector.collect()}
        self.assertEqual(
            metrics["test"].samples,
            [
                Sample("test_total", {"label": "a"}, 1.0),
                Sample("test_total", {"label": "b"}, 7.0),
            ],
        )

    def test_many_metrics(self) -> None:
        counter = Counter(
            "test",
            "Test Counter",
            labelnames=("label",),
            registry=self.registry,
        )
        for label in range(100):
            counter.labels(label=str(label)).inc()
        metrics = {m.name: m for m in self.collector.collect()}
        self.assertEqual(len(metrics["test"].samples), 100)

    def test_histogram(self) -> None:
        histogram = Histogram(
            "test",
            "Test Histogram",
            buckets=(0.01, 0.1, 1, 10, float("inf")),
            registry=self.registry,
        )
        histogram.observe(0.1)
        histogram.observe(0.5)
        histogram.observe(5)
        histogram.observe(50)
        metrics = {m.name: m for m in self.collector.collect()}
        self.assertEqual(
            metrics["test"].samples,
            [
                Sample("test_bucket", {"le": "0.01"}, 0.0),
                Sample("test_bucket", {"le": "0.1"}, 1.0),
                Sample("test_bucket", {"le": "1.0"}, 2.0),
                Sample("test_bucket", {"le": "10.0"}, 3.0),
                Sample("test_bucket", {"le": "+Inf"}, 4.0),
                Sample("test_count", {}, 4.0),
                Sample("test_sum", {}, 55.6),
            ],
        )

    def test_gauge_all(self) -> None:
        g1 = Gauge("g", "help", registry=None)
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None)
        self.assertEqual(0, self.registry.get_sample_value("g", {"pid": "123"}))
        self.assertEqual(0, self.registry.get_sample_value("g", {"pid": "456"}))
        g1.set(1)
        g2.set(2)
        mark_process_dead(123)
        self.assertEqual(1, self.registry.get_sample_value("g", {"pid": "123"}))
        self.assertEqual(2, self.registry.get_sample_value("g", {"pid": "456"}))

    def test_gauge_liveall(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="liveall")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="liveall")
        self.assertEqual(0, self.registry.get_sample_value("g", {"pid": "123"}))
        self.assertEqual(0, self.registry.get_sample_value("g", {"pid": "456"}))
        g1.set(1)
        g2.set(2)
        self.assertEqual(1, self.registry.get_sample_value("g", {"pid": "123"}))
        self.assertEqual(2, self.registry.get_sample_value("g", {"pid": "456"}))
        mark_process_dead(123)
        self.assertEqual(
            None, self.registry.get_sample_value("g", {"pid": "123"})
        )
        self.assertEqual(2, self.registry.get_sample_value("g", {"pid": "456"}))

    def test_gauge_min(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="min")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="min")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(1)
        g2.set(2)
        self.assertEqual(1, self.registry.get_sample_value("g"))

    def test_gauge_livemin(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="livemin")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="livemin")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(1)
        g2.set(2)
        self.assertEqual(1, self.registry.get_sample_value("g"))
        mark_process_dead(123)
        self.assertEqual(2, self.registry.get_sample_value("g"))

    def test_gauge_max(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="max")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="max")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(1)
        g2.set(2)
        self.assertEqual(2, self.registry.get_sample_value("g"))

    def test_gauge_livemax(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="livemax")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="livemax")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(2)
        g2.set(1)
        self.assertEqual(2, self.registry.get_sample_value("g"))
        mark_process_dead(123)
        self.assertEqual(1, self.registry.get_sample_value("g"))

    def test_gauge_sum(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="sum")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="sum")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(1)
        g2.set(2)
        self.assertEqual(3, self.registry.get_sample_value("g"))
        mark_process_dead(123)
        self.assertEqual(3, self.registry.get_sample_value("g"))

    def test_gauge_livesum(self) -> None:
        g1 = Gauge("g", "help", registry=None, multiprocess_mode="livesum")
        values.ValueClass = RedisValue(lambda: 456)
        g2 = Gauge("g", "help", registry=None, multiprocess_mode="livesum")
        self.assertEqual(0, self.registry.get_sample_value("g"))
        g1.set(1)
        g2.set(2)
        self.assertEqual(3, self.registry.get_sample_value("g"))
        mark_process_dead(123)
        self.assertEqual(2, self.registry.get_sample_value("g"))
