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

import time as tm
from threading import Event
from typing import override
from unittest import mock

from django.test import SimpleTestCase, override_settings
from fakeredis import FakeRedis
from redis import Redis

from debusine.server.prometheus_client.redis import (
    _daemon_threads,
    _keep_key_from_expiring,
    mark_process_dead,
    private_fake_redis,
    redis_client,
)


class RedisClientTestCase(SimpleTestCase):
    """Test redis_client()."""

    @override_settings(PROMETHEUS_REDIS_URL="redis://localhost/0")
    def test_redis(self) -> None:
        client = redis_client()
        self.assertIsInstance(client, Redis)
        self.assertNotIsInstance(client, FakeRedis)

    @private_fake_redis()
    def test_fake_redis(self) -> None:
        client = redis_client()
        self.assertIsInstance(client, FakeRedis)


@private_fake_redis()
class KeepMetricsAliveTestCase(SimpleTestCase):
    """Test KeepMetricsAliveThread and friends."""

    @override
    def setUp(self) -> None:
        super().setUp()
        self.stop_event = Event()

    @override
    def tearDown(self) -> None:
        super().tearDown()
        self.stop_event.set()

    def mock_loop(self) -> None:
        self.loop_event = Event()
        self.waiting_event = Event()
        patcher = mock.patch(
            (
                "debusine.server.prometheus_client.redis."
                "KeepMetricsAliveThread.loop_wait"
            ),
            side_effect=self.loop_wait,
        )
        self.addCleanup(patcher.stop)
        patcher.start()

    def loop_wait(self, _timeout: float) -> bool:
        self.waiting_event.set()
        self.waiting_event = Event()
        assert self.loop_event.wait(5) is True
        return self.stop_event.is_set()

    def run_loop_once(self, stop: bool = False) -> None:
        if stop:
            self.stop_event.set()
        self.loop_event.set()
        if not stop:
            self.loop_event = Event()
            assert self.waiting_event.wait() is True

    def test_mark_unknown_process_dead(self) -> None:
        mark_process_dead("unknown")

    def test_mark_known_process_dead(self) -> None:
        client = redis_client()
        client.set("key", "hello")
        _keep_key_from_expiring("identifier", "key")
        self.assertIn("identifier", _daemon_threads)
        mark_process_dead("identifier")
        self.assertNotIn("identifier", _daemon_threads)

    def test_live_daemon_updates_expiry(self) -> None:
        """Exercise the full lifecycle of the daemon thread."""
        self.mock_loop()
        client = redis_client()
        client.set("key", "hello")
        _keep_key_from_expiring("identifier", "key")
        self.assertLess(client.expiretime("key"), 0)

        unixtime = tm.time()
        self.run_loop_once()
        self.assertGreater(client.expiretime("key"), unixtime)
        self.run_loop_once(stop=True)
