python-repositories/tests/unit/connection_aware_adapter_test.py

"""Unit tests for TTL-cached connection health checks on ConnectionAwareAdapter."""

from __future__ import annotations

from typing import cast
from unittest.mock import MagicMock, patch

import redis

from python_repositories.adapters.minio_adapter import MinioAdapter
from python_repositories.adapters.redis_adapter import RedisAdapter
from tests.conftest import TEST_MINIO_CONFIG, TEST_REDIS_CONFIG


class TestRedisConnectionHealth:
    def test_not_connected_when_no_client(self) -> None:
        adapter = RedisAdapter(config=TEST_REDIS_CONFIG)
        assert not adapter.is_connected()

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

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

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

        assert not redis_adapter.is_connected()

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

        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 = cast(MagicMock, redis_adapter._client)
        mock_client.ping.return_value = True

        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 = cast(MagicMock, redis_adapter._client)
        mock_client.ping.return_value = True

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

        redis_adapter.disconnect()
        reinjected = RedisAdapter(
            config=redis_adapter._config,
            client=mock_client,
        )

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

        assert mock_client.ping.call_count == 2


class TestMinioConnectionHealth:
    def test_not_connected_when_no_client(self) -> None:
        adapter = MinioAdapter(config=TEST_MINIO_CONFIG)
        assert not adapter.is_connected()

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

        assert not minio_adapter.is_connected()

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

        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:
        cast(MagicMock, minio_adapter._client).bucket_exists.side_effect = Exception(
            "connection lost"
        )

        assert not minio_adapter.is_connected()

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

        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 = cast(MagicMock, minio_adapter._client)
        mock_client.bucket_exists.return_value = True

        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 = cast(MagicMock, minio_adapter._client)
        mock_client.bucket_exists.return_value = True

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

        minio_adapter.disconnect()
        reinjected = MinioAdapter(
            config=minio_adapter._config,
            client=mock_client,
        )

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

        assert mock_client.bucket_exists.call_count == 2