Brian Bjarke JensenandCursor
beb2b5128e
Code Quality Pipeline / code-quality (pull_request) Failing after 29s
Test Python Package / integration-tests (pull_request) Successful in 35s
Test Python Package / unit-tests (pull_request) Successful in 42s
PR Title Check / check-title (pull_request) Successful in 52s
Test Python Package / coverage-report (pull_request) Successful in 12s
Add fast unit tests for lazy-import helpers and injected-client connect paths that integration tests miss, and raise the coverage floor to 100%. Co-authored-by: Cursor <[email protected]>
117 lines
3.8 KiB
Python
117 lines
3.8 KiB
Python
"""Unit tests for RedisAdapter instantiation and injection."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
import redis
|
|
|
|
from python_repositories.adapters.redis_adapter import RedisAdapter
|
|
from python_repositories.interfaces import JsonRepositoryInterface
|
|
from tests.conftest import TEST_REDIS_CONFIG
|
|
|
|
|
|
def test_should_adhere_to_interface() -> None:
|
|
assert issubclass(RedisAdapter, JsonRepositoryInterface)
|
|
_ = RedisAdapter(config=TEST_REDIS_CONFIG)
|
|
|
|
|
|
def test_should_have_logger_when_instantiated() -> None:
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG)
|
|
assert hasattr(adapter, "logger")
|
|
assert adapter.logger is not None
|
|
|
|
|
|
def test_should_not_be_connected_when_instantiated() -> None:
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG)
|
|
assert adapter._client is None
|
|
assert not adapter.is_connected()
|
|
|
|
|
|
def test_constructs_with_injected_config_without_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.delenv("REDIS_URI", raising=False)
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG)
|
|
assert adapter._config == TEST_REDIS_CONFIG
|
|
|
|
|
|
def test_constructs_with_injected_client_without_env(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
monkeypatch.delenv("REDIS_URI", raising=False)
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG, client=mock_client)
|
|
assert adapter._client is mock_client
|
|
assert adapter._client_injected is True
|
|
|
|
|
|
def test_raises_when_client_provided_without_config() -> None:
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
with pytest.raises(ValueError, match="config is required"):
|
|
RedisAdapter(client=mock_client)
|
|
|
|
|
|
def test_disconnect_does_not_close_injected_client() -> None:
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG, client=mock_client)
|
|
adapter.disconnect()
|
|
mock_client.close.assert_not_called()
|
|
assert adapter._client is None
|
|
|
|
|
|
class CustomEnvRedisAdapter(RedisAdapter):
|
|
uri_env_var_name = "CUSTOM_REDIS_URI"
|
|
|
|
|
|
def test_subclass_custom_env_var_name(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("CUSTOM_REDIS_URI", "redis://custom:6379")
|
|
adapter = CustomEnvRedisAdapter()
|
|
assert adapter._config.uri == "redis://custom:6379"
|
|
|
|
|
|
def test_connect_with_injected_client_succeeds_when_ping_ok() -> None:
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
mock_client.ping.return_value = True
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG, client=mock_client)
|
|
|
|
adapter.connect()
|
|
|
|
mock_client.ping.assert_called_once()
|
|
|
|
|
|
def test_connect_with_injected_client_raises_when_ping_false() -> None:
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
mock_client.ping.return_value = False
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG, client=mock_client)
|
|
|
|
with pytest.raises(ConnectionError, match="Could not connect to Redis"):
|
|
adapter.connect()
|
|
|
|
|
|
def test_connect_with_injected_client_raises_on_redis_error() -> None:
|
|
mock_client = MagicMock(spec=redis.Redis)
|
|
mock_client.ping.side_effect = redis.ConnectionError("connection lost")
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG, client=mock_client)
|
|
|
|
with pytest.raises(ConnectionError, match="Could not connect to Redis"):
|
|
adapter.connect()
|
|
|
|
|
|
def test_connect_closes_existing_non_injected_client(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
stale_client = MagicMock(spec=redis.Redis)
|
|
new_client = MagicMock(spec=redis.Redis)
|
|
new_client.ping.return_value = True
|
|
adapter = RedisAdapter(config=TEST_REDIS_CONFIG)
|
|
adapter._client = stale_client
|
|
|
|
monkeypatch.setattr("redis.Redis.from_url", lambda *args, **kwargs: new_client)
|
|
|
|
adapter.connect()
|
|
|
|
stale_client.close.assert_called_once()
|
|
assert adapter._client is new_client
|