python-repositories/tests/unit/connection_health_test.py

"""Tests for TTL-cached connection health checks on adapters."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

import pytest
import redis

from python_repositories.adapters.minio_adapter import MinioAdapter
from python_repositories.adapters.redis_adapter import RedisAdapter


@pytest.fixture
def redis_adapter(monkeypatch: pytest.MonkeyPatch) -> RedisAdapter:
    monkeypatch.setenv("REDIS_URI", "redis://localhost:6379")
    return RedisAdapter()


@pytest.fixture
def minio_adapter(monkeypatch: pytest.MonkeyPatch) -> MinioAdapter:
    monkeypatch.setenv("MINIO_ENDPOINT", "localhost:9000")
    monkeypatch.setenv("MINIO_ACCESS_KEY", "minioadmin")
    monkeypatch.setenv("MINIO_SECRET_KEY", "minioadmin")
    monkeypatch.setenv("MINIO_BUCKET", "test-bucket")
    return MinioAdapter()


class TestRedisConnectionHealth:
    def test_not_connected_when_no_client(self, redis_adapter: RedisAdapter) -> None:
        assert not redis_adapter.is_connected()

    def test_connected_when_probe_succeeds(self, redis_adapter: RedisAdapter) -> None:
        mock_client = MagicMock(spec=redis.Redis)
        mock_client.ping.return_value = True
        redis_adapter._client = mock_client

        assert redis_adapter.is_connected()
        mock_client.ping.assert_called_once()

    def test_stale_connection_when_probe_fails(
        self, redis_adapter: RedisAdapter
    ) -> None:
        mock_client = MagicMock(spec=redis.Redis)
        mock_client.ping.side_effect = redis.ConnectionError("connection lost")
        redis_adapter._client = mock_client

        assert not redis_adapter.is_connected()

    def test_cache_hit_avoids_second_probe(self, redis_adapter: RedisAdapter) -> None:
        mock_client = MagicMock(spec=redis.Redis)
        mock_client.ping.return_value = True
        redis_adapter._client = mock_client

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert redis_adapter.is_connected()
            assert redis_adapter.is_connected()

        mock_client.ping.assert_called_once()

    def test_cache_miss_runs_probe_again(self, redis_adapter: RedisAdapter) -> None:
        mock_client = MagicMock(spec=redis.Redis)
        mock_client.ping.return_value = True
        redis_adapter._client = mock_client

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            side_effect=[100.0, 102.0],
        ):
            assert redis_adapter.is_connected()
            assert redis_adapter.is_connected()

        assert mock_client.ping.call_count == 2

    def test_disconnect_clears_cache(self, redis_adapter: RedisAdapter) -> None:
        mock_client = MagicMock(spec=redis.Redis)
        mock_client.ping.return_value = True
        redis_adapter._client = mock_client

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert redis_adapter.is_connected()

        redis_adapter.disconnect()
        redis_adapter._client = mock_client

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert redis_adapter.is_connected()

        assert mock_client.ping.call_count == 2


class TestMinioConnectionHealth:
    def test_not_connected_when_no_client(self, minio_adapter: MinioAdapter) -> None:
        assert not minio_adapter.is_connected()

    def test_not_connected_when_bucket_name_missing(
        self, minio_adapter: MinioAdapter
    ) -> None:
        minio_adapter._client = MagicMock()
        minio_adapter._bucket_name = None

        assert not minio_adapter.is_connected()

    def test_connected_when_probe_succeeds(self, minio_adapter: MinioAdapter) -> None:
        mock_client = MagicMock()
        mock_client.bucket_exists.return_value = True
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        assert minio_adapter.is_connected()
        mock_client.bucket_exists.assert_called_once_with("test-bucket")

    def test_stale_connection_when_probe_fails(
        self, minio_adapter: MinioAdapter
    ) -> None:
        mock_client = MagicMock()
        mock_client.bucket_exists.side_effect = Exception("connection lost")
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        assert not minio_adapter.is_connected()

    def test_cache_hit_avoids_second_probe(self, minio_adapter: MinioAdapter) -> None:
        mock_client = MagicMock()
        mock_client.bucket_exists.return_value = True
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert minio_adapter.is_connected()
            assert minio_adapter.is_connected()

        mock_client.bucket_exists.assert_called_once()

    def test_cache_miss_runs_probe_again(self, minio_adapter: MinioAdapter) -> None:
        mock_client = MagicMock()
        mock_client.bucket_exists.return_value = True
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            side_effect=[100.0, 102.0],
        ):
            assert minio_adapter.is_connected()
            assert minio_adapter.is_connected()

        assert mock_client.bucket_exists.call_count == 2

    def test_disconnect_clears_cache(self, minio_adapter: MinioAdapter) -> None:
        mock_client = MagicMock()
        mock_client.bucket_exists.return_value = True
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert minio_adapter.is_connected()

        minio_adapter.disconnect()
        minio_adapter._client = mock_client
        minio_adapter._bucket_name = "test-bucket"

        with patch(
            "python_repositories.adapters.connection_aware_adapter.time.monotonic",
            return_value=100.0,
        ):
            assert minio_adapter.is_connected()

        assert mock_client.bucket_exists.call_count == 2