"""Tests for Redis Connection methods.""" import logging from typing import Any, Dict, List, Optional, TypeVar from unittest.mock import AsyncMock, MagicMock, call, patch import pytest import redis.asyncio as redis from redis.exceptions import RedisError from pdp import config from pdp.connectors.redis_client import ( JSONSerializer, PPCacheItemSerializer, RedisConnector, RedisConnectorFactoryError, redis_connector_factory, ) from tests.unit.helpers import WithAsyncIterator, awaitable_return_value T = TypeVar("T") def to_string_array(keys_return_value: List[bytes]) -> List[str]: """Convert byte string to a normal Python string.""" if not keys_return_value: return [] return [str(k, encoding="utf-8") for k in keys_return_value] @patch("pdp.connectors.redis_client.redis") def test_init(mock_redis_module: MagicMock) -> None: """Test redis client initialization.""" _ = RedisConnector(redis_url="no.such.host", use_redis_cache=True) mock_redis_module.Redis.from_url.assert_called_with( url="no.such.host", socket_connect_timeout=config.REDIS_SOCKET_CONNECT_TIMEOUT, socket_timeout=config.REDIS_SOCKET_TIMEOUT, ) async def test_ping( mock_redis_connector: MagicMock, mock_redis_client: MagicMock ) -> None: """Test the ping method.""" mock_redis_client.ping.return_value = awaitable_return_value("dog") redis_response = await mock_redis_connector.ping() assert redis_response is True mock_redis_client.ping.assert_called_once() async def test_ping_no_connection( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test the ping method when there is a RedisError.""" mock_redis_client.ping.side_effect = RedisError("TimeoutError or ConnectionError") redis_response = await mock_redis_connector.ping() assert redis_response is False mock_redis_client.ping.assert_called_once() assert ( caplog.records[0].msg == "Failed to ping Redis at no.such.host with use_redis_cache = False" ) async def test_dbsize( mock_redis_connector: MagicMock, mock_redis_client: MagicMock ) -> None: """Test the dbsize method.""" mock_redis_client.dbsize.return_value = awaitable_return_value(10) redis_response = await mock_redis_connector.dbsize() assert redis_response == 10 mock_redis_client.dbsize.assert_called_once() @pytest.mark.parametrize( "cache_key, cached_objects, serializer", [ pytest.param( "foo#identity_uuid", {"key": "value"}, JSONSerializer(), id="GET should return a Python dictionary object", ), pytest.param( "no.such.key", None, JSONSerializer(), id="GET should handle None from the redis client.", ), ], ) async def test_get( cache_key: str, cached_objects: T, serializer: PPCacheItemSerializer, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the GET method.""" mock_redis_client.mget.return_value = awaitable_return_value( [serializer.dump(cached_objects)] ) redis_response = await mock_redis_connector.get(cache_key, serializer=serializer) assert redis_response == cached_objects mock_redis_client.mget.assert_called_with(keys=[cache_key]) @pytest.mark.parametrize( "object_to_cache, set_return_value, serializer", [ pytest.param( {"key": "value"}, True, JSONSerializer(), id="SET should succeed with a JSONSerializer", ), pytest.param( None, False, JSONSerializer(), id="SET should not fail when called with null", ), ], ) async def test_set( object_to_cache: T, set_return_value: bool, serializer: PPCacheItemSerializer, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Tests the SET command.""" mock_redis_client.set.return_value = awaitable_return_value(set_return_value) redis_response = await mock_redis_connector.set( "cache_key", object_to_cache, serializer=serializer ) assert redis_response == set_return_value if object_to_cache is None: mock_redis_client.set.assert_not_called() else: mock_redis_client.set.assert_called_with( "cache_key", serializer.dump(object_to_cache), ex=config.REDIS_CACHE_TTL ) async def test_set_uses_override_redis_client( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Tests a different redis_client can be used instead of object's client.""" alternate_redis_client = AsyncMock(spec=redis.Redis) alternate_redis_client.set.return_value = awaitable_return_value(True) await mock_redis_connector.set( "cache_key", {"key": "value"}, serializer=JSONSerializer(), redis_client=alternate_redis_client, ) mock_redis_client.set.assert_not_called() alternate_redis_client.set.assert_called_once_with( "cache_key", '{"key": "value"}', ex=config.REDIS_CACHE_TTL, ) @pytest.mark.parametrize( "ttl", [ pytest.param(None, id="None means no expiration"), pytest.param(123, id="ttl can be a number"), ], ) async def test_set_ttl( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ttl: Optional[int], ) -> None: """Tests the SET command handles ttl.""" object_to_cache = {"some": "object"} serializer = JSONSerializer() mock_redis_client.set.return_value = awaitable_return_value(True) redis_response = await mock_redis_connector.set( "cache_key", object_to_cache, serializer=serializer, ttl=ttl ) assert redis_response mock_redis_client.set.assert_called_with( "cache_key", serializer.dump(object_to_cache), ex=ttl, ) async def test_set_redis_error( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test set handles RedisError.""" key = "some-key" item = {"some": "value"} mock_redis_client.set.side_effect = RedisError("set redis error") redis_response = await mock_redis_connector.set( key, item, serializer=JSONSerializer() ) assert not redis_response mock_redis_client.set.assert_called_once_with( key, '{"some": "value"}', ex=config.REDIS_CACHE_TTL, ) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, "RedisError should be logged" assert caplog.records[0].msg == "[RedisError] set failed, returning False" @pytest.mark.parametrize( "key_value_objects_to_cache", [ pytest.param( { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, }, id="MSET should SET each item in the dictionary", ), ], ) async def test_mset( key_value_objects_to_cache: Dict[str, Any], mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache MSET method.""" serializer = JSONSerializer() mock_redis_client.set.return_value = awaitable_return_value(return_value=True) redis_response = await mock_redis_connector.mset( key_item_dict=key_value_objects_to_cache, serializer=serializer ) assert redis_response == dict(zip(key_value_objects_to_cache.keys(), [True, True])) expected_calls = [ call(key, serializer.dump(value), ex=config.REDIS_CACHE_TTL) for key, value in key_value_objects_to_cache.items() ] # This is asserting the calls made to the Redis client SET method mock_redis_client.set.assert_has_calls(expected_calls) @pytest.mark.parametrize( "ttl", [ pytest.param(None, id="None means no expiration"), pytest.param(123, id="ttl can be a number"), ], ) async def test_mset_ttl( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ttl: Optional[int], ) -> None: """Test the cache MSET method handles ttl.""" key_value_objects_to_cache = {"key": {"some": "item"}} serializer = JSONSerializer() mock_redis_client.set.return_value = awaitable_return_value(return_value=True) redis_response = await mock_redis_connector.mset( key_item_dict=key_value_objects_to_cache, serializer=serializer, ttl=ttl, ) assert redis_response == dict(zip(key_value_objects_to_cache.keys(), [True])) expected_calls = [ call(key, serializer.dump(value), ex=ttl) for key, value in key_value_objects_to_cache.items() ] # This is asserting the calls made to the Redis client SET method mock_redis_client.set.assert_has_calls(expected_calls) @pytest.mark.parametrize( "redis_side_effects, expected_response", [ pytest.param( [RedisError("set redis error"), RedisError("set redis error")], [False, False], id="MSET should handle RedisError", ), pytest.param( [True, RedisError("set redis error")], [True, False], id="MSET should handle RedisError and success", ), ], ) async def test_mset_error( redis_side_effects: List[Any], expected_response: List[bool], mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test mset handles RedisError.""" key_value_objects_to_cache = { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, } serializer = JSONSerializer() mock_redis_client.set = AsyncMock(side_effect=redis_side_effects) redis_response = await mock_redis_connector.mset( key_item_dict=key_value_objects_to_cache, serializer=serializer ) assert redis_response == dict( zip(key_value_objects_to_cache.keys(), expected_response) ) expected_calls = [ call(key, serializer.dump(value), ex=config.REDIS_CACHE_TTL) for key, value in key_value_objects_to_cache.items() ] mock_redis_client.set.assert_has_calls(expected_calls) with caplog.at_level(logging.ERROR): for record in caplog.records: assert record.msg == "[RedisError] set failed, returning False" @pytest.mark.parametrize( "key_value_objects_to_cache", [ pytest.param( { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, }, id="MSET should SET each item in the dictionary", ), ], ) async def test_mset_with_pipeline( key_value_objects_to_cache: Dict[str, Any], mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache MSET with pipeline method.""" serializer = JSONSerializer() mock_pipeline = MagicMock(spec=redis.client.Pipeline) mock_pipeline.execute = AsyncMock(return_value=[True, True]) mock_pipeline.reset.return_value = AsyncMock(return_value=None) mock_redis_client.pipeline.return_value = mock_pipeline mock_redis_connector.set = AsyncMock() actual = await mock_redis_connector.mset_with_pipeline( key_item_dict=key_value_objects_to_cache, serializer=serializer ) assert actual == dict(zip(key_value_objects_to_cache.keys(), [True, True])) expected_calls = [ call( key, value, serializer=serializer, ttl=config.REDIS_CACHE_TTL, redis_client=mock_pipeline, ) for key, value in key_value_objects_to_cache.items() ] mock_redis_connector.set.assert_has_calls(expected_calls) mock_redis_client.pipeline.assert_called_once_with(transaction=False) mock_pipeline.execute.assert_called_once_with(raise_on_error=True) mock_pipeline.reset.assert_called_once() @pytest.mark.parametrize( "ttl", [ pytest.param(None, id="None means no expiration"), pytest.param(123, id="ttl can be a number"), ], ) async def test_mset_with_pipeline_ttl( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ttl: Optional[int], ) -> None: """Test the cache MSET with pipeline method handles ttl.""" key_value_objects_to_cache = {"key": {"some": "item"}} serializer = JSONSerializer() mock_pipeline = MagicMock(spec=redis.client.Pipeline) mock_pipeline.execute = AsyncMock(return_value=[True]) mock_pipeline.reset.return_value = AsyncMock(return_value=None) mock_redis_client.pipeline.return_value = mock_pipeline mock_redis_connector.set = AsyncMock() actual = await mock_redis_connector.mset_with_pipeline( key_item_dict=key_value_objects_to_cache, serializer=serializer, ttl=ttl, ) assert actual == dict(zip(key_value_objects_to_cache.keys(), [True])) expected_calls = [ call(key, value, serializer=serializer, ttl=ttl, redis_client=mock_pipeline) for key, value in key_value_objects_to_cache.items() ] # This is asserting the calls made to the Redis client SET method mock_redis_connector.set.assert_has_calls(expected_calls) mock_redis_client.pipeline.assert_called_once_with(transaction=False) mock_pipeline.execute.assert_called_once_with(raise_on_error=True) mock_pipeline.reset.assert_called_once() @pytest.mark.parametrize( "execute_return_value, expected_response, expected_log", [ pytest.param( [RedisError("set redis error")], [False, False], "[RedisError] Pipeline Execute Failed, returning False", id="Pipeline execute error", ), pytest.param( [ [False], ], [False, False], "[AssertionError] received mismatched result length", id="Execute return array is different length from requested dict", ), ], ) async def test_mset_with_pipeline_error( execute_return_value: Any, expected_response: List[bool], expected_log: str, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test mset with pipeline handles error scenarios.""" key_value_objects_to_cache = { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, } serializer = JSONSerializer() mock_pipeline = MagicMock(spec=redis.client.Pipeline) mock_pipeline.execute = AsyncMock(side_effect=execute_return_value) mock_pipeline.reset.return_value = AsyncMock(return_value=None) mock_redis_client.pipeline.return_value = mock_pipeline mock_redis_connector.set = AsyncMock() actual = await mock_redis_connector.mset_with_pipeline( key_item_dict=key_value_objects_to_cache, serializer=serializer ) assert actual == dict(zip(key_value_objects_to_cache.keys(), expected_response)) expected_calls = [ call( key, value, serializer=serializer, ttl=config.REDIS_CACHE_TTL, redis_client=mock_pipeline, ) for key, value in key_value_objects_to_cache.items() ] mock_redis_connector.set.assert_has_calls(expected_calls) mock_redis_client.pipeline.assert_called_once_with(transaction=False) mock_pipeline.execute.assert_called_once_with(raise_on_error=True) mock_pipeline.reset.assert_called_once() with caplog.at_level(logging.ERROR): for record in caplog.records: assert record.msg == expected_log @pytest.mark.parametrize( "key_value_objects_to_cache, expected_return_value, serializer", [ pytest.param( { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, }, [{"key1": "value1"}, {"key2": "value2"}], JSONSerializer(), id="MGET should return an Python dictionary object", ), pytest.param( { "foo#identity_uuid_01": None, "foo#identity_uuid_02": None, }, [None, None], JSONSerializer(), id="MGET should handle cache misses", ), pytest.param( {}, [], JSONSerializer(), id="MGET should handle empty list from the redis client.", ), ], ) async def test_mget( key_value_objects_to_cache: Dict[str, Any], expected_return_value: List[Any], serializer: PPCacheItemSerializer, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache SET and MGET methods.""" keys = [k for k in key_value_objects_to_cache.keys()] mock_redis_client.mget.return_value = awaitable_return_value( [serializer.dump(item) for item in key_value_objects_to_cache.values()] ) redis_response = await mock_redis_connector.mget(keys, serializer=serializer) assert redis_response == expected_return_value if expected_return_value: mock_redis_client.mget.assert_called_with(keys=keys) else: mock_redis_client.mget.assert_not_called() async def test_mget_redis_error( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test mget handles RedisError.""" keys = ["does", "not", "matter"] mock_redis_client.mget.side_effect = RedisError("mget redis error") redis_response = await mock_redis_connector.mget(keys, serializer=JSONSerializer()) assert redis_response == [None, None, None] mock_redis_client.mget.assert_called_once_with(keys=keys) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, "RedisError should be logged" assert ( caplog.records[0].msg == "[RedisError] mget failed, returning array of Nones" ) @pytest.mark.parametrize( "exception_to_raise, expected_log_str", [ pytest.param( ValueError(), "Load failed for key", id="MGET should return None for ValueError", ), pytest.param( Exception(), "Unexpected load error", id="MGET should return None for Exception", ), ], ) async def test_mget_load_errors( exception_to_raise: Exception, expected_log_str: str, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test the cache SET and MGET methods with load errors.""" keys = ["key1", "key2"] values = ["value1", "value2"] expected_response = [None, None] mock_redis_client.mget.return_value = awaitable_return_value(values) serializer = MagicMock(spec=PPCacheItemSerializer) serializer.load.side_effect = exception_to_raise redis_response = await mock_redis_connector.mget(keys, serializer=serializer) assert redis_response == expected_response mock_redis_client.mget.assert_called_with(keys=keys) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 2 for record in caplog.records: assert expected_log_str in record.msg @pytest.mark.parametrize( "delete_return_value", [ pytest.param( 2, id="Delete should succeed when redis returns non-zero", ), pytest.param( 0, id="Delete should succeed when redis returns zero.", ), ], ) async def test_delete( delete_return_value: int, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache DELETE method.""" mock_redis_client.delete.return_value = awaitable_return_value(delete_return_value) redis_response = await mock_redis_connector.delete("prefix_key1", "prefix_key2") assert redis_response == delete_return_value mock_redis_client.delete.assert_called_with("prefix_key1", "prefix_key2") async def test_delete_redis_error( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test delete handles RedisError.""" keys = ["does", "not", "matter"] mock_redis_client.delete.side_effect = RedisError("delete redis error") redis_response = await mock_redis_connector.delete(keys) assert redis_response == 0 mock_redis_client.delete.assert_called_once_with(keys) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, "RedisError should be logged" assert caplog.records[0].msg == "[RedisError] delete failed, returning 0" @pytest.mark.parametrize( "keys_return_value", [ pytest.param( [b"prefix_key1", b"prefix_key2"], id="List should succeed when keys returns non-empty list", ), pytest.param( [], id="List should succeed when keys returns empty list", ), ], ) async def test_list( keys_return_value: List[bytes], mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache list method.""" mock_redis_client.scan_iter.return_value = WithAsyncIterator( items=keys_return_value ) redis_response = await mock_redis_connector.list("prefix_*") assert redis_response == to_string_array(keys_return_value) mock_redis_client.scan_iter.assert_called_with(match="prefix_*") async def test_list_redis_error( mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: Any, ) -> None: """Test list handles RedisError.""" mock_redis_client.scan_iter.side_effect = RedisError("scan_iter redis error") redis_response = await mock_redis_connector.list("prefix_*") assert redis_response == [] mock_redis_client.scan_iter.assert_called_once_with(match="prefix_*") with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, "RedisError should be logged" assert caplog.records[0].msg == "[RedisError] scan_iter failed, returning []" @pytest.mark.parametrize( "delete_return_value,keys_return_value,expect_error", [ pytest.param( 2, [b"prefix_key1", b"prefix_key2"], False, id="The method should succeed when deleting entries", ), pytest.param( 0, [], False, id="The method should not call delete if 0 keys returned.", ), pytest.param( 1, [b"prefix_key1", b"prefix_key2"], True, id="""The method should log a warning because DELETE and SCAN count values don't match""", # noqa: E501 ), ], ) async def test_delete_all_matching_pattern( delete_return_value: int, keys_return_value: List[Any], expect_error: bool, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test the cache list method.""" mock_redis_client.scan_iter.return_value = WithAsyncIterator( items=keys_return_value ) mock_redis_client.delete.return_value = awaitable_return_value(delete_return_value) expected_response = to_string_array(keys_return_value) redis_response = await mock_redis_connector.delete_all_matching_pattern("prefix_*") assert redis_response == expected_response mock_redis_client.scan_iter.assert_called_with(match="prefix_*") if keys_return_value: mock_redis_client.delete.assert_called_with(*expected_response) else: mock_redis_client.delete.assert_not_called() if expect_error: with caplog.at_level(logging.WARNING): assert "deleted keys but DELETE returned" in caplog.text @pytest.mark.parametrize( "key_value_objects_to_cache, expected_return_value, serializer", [ pytest.param( { "foo#identity_uuid_01": {"key1": "value1"}, "foo#identity_uuid_02": {"key2": "value2"}, }, [{"key1": "value1"}, {"key2": "value2"}], JSONSerializer(), id="GETDEL should return an Python dictionary object", ), pytest.param( { "foo#identity_uuid_01": None, "foo#identity_uuid_02": None, }, [None, None], JSONSerializer(), id="GETDEL should handle cache misses", ), pytest.param( {}, [], JSONSerializer(), id="GETDEL should handle empty list from the redis client.", ), ], ) async def test_getdel( key_value_objects_to_cache: Dict[str, Any], expected_return_value: List[Any], serializer: PPCacheItemSerializer, mock_redis_connector: MagicMock, mock_redis_client: MagicMock, ) -> None: """Test the cache GETDEL methods.""" keys = [k for k in key_value_objects_to_cache.keys()] mock_redis_client.getdel.side_effect = [ awaitable_return_value(serializer.dump(item)) for item in key_value_objects_to_cache.values() ] redis_response = await mock_redis_connector.getdel(keys, serializer=serializer) assert redis_response == expected_return_value if expected_return_value: print(f"Calls: {mock_redis_client.getdel.mock_calls}") expected_calls = [call(name=k) for k in keys] mock_redis_client.getdel.assert_has_calls(expected_calls) else: mock_redis_client.getdel.assert_not_called() @patch("pdp.connectors.redis_client.RedisConnector") @patch("pdp.connectors.redis_client.config") async def test_redis_connector_factory( mock_config: MagicMock, mock_redis_connector_obj: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test RedisConnector is created, happy path.""" mock_config.REDIS_URL = "rediss://host:port" mock_config.CACHE_USE_REDIS = True mock_redis_connector = MagicMock(spec=RedisConnector) mock_redis_connector.ping.return_value = True mock_redis_connector_obj.return_value = mock_redis_connector with caplog.at_level(logging.INFO): actual = await redis_connector_factory() assert actual == mock_redis_connector mock_redis_connector_obj.assert_called_once() mock_redis_connector.ping.assert_called_once() assert caplog.records[0].levelno == logging.INFO assert ( caplog.records[0].message == "Initialized RedisConnector at rediss://host:port with use_redis_cache = True" # noqa: E501 ) @patch("pdp.connectors.redis_client.RedisConnector") @patch("pdp.connectors.redis_client.config") async def test_redis_connector_factory_fallback( mock_config: MagicMock, mock_redis_connector_obj: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test RedisConnector is created using fakeredis as fallback.""" mock_config.REDIS_URL = "rediss://host:port" mock_config.CACHE_USE_REDIS = True mock_redis_connector = MagicMock(spec=RedisConnector) mock_redis_connector.ping.return_value = False mock_fakeredis_connector = MagicMock(spec=RedisConnector) mock_fakeredis_connector.ping.return_value = True mock_redis_connector_obj.side_effect = [ mock_redis_connector, mock_fakeredis_connector, ] with caplog.at_level(logging.INFO): actual = await redis_connector_factory() assert actual == mock_fakeredis_connector assert mock_redis_connector_obj.call_count == 2 mock_redis_connector.ping.assert_called_once() mock_fakeredis_connector.ping.assert_called_once() assert caplog.records[0].levelno == logging.ERROR assert caplog.records[0].message == "Switching to use fakeredis." assert caplog.records[1].levelno == logging.INFO assert caplog.records[1].message == "Initialized RedisConnector with fakeredis" @patch("pdp.connectors.redis_client.RedisConnector") @patch("pdp.connectors.redis_client.config") async def test_redis_connector_factory_fallback_error( mock_config: MagicMock, mock_redis_connector_obj: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test redis_connector_factory raises RedisConnectorFactoryError.""" mock_config.REDIS_URL = "rediss://host:port" mock_config.CACHE_USE_REDIS = True mock_redis_connector = MagicMock(spec=RedisConnector) mock_redis_connector.ping.return_value = False mock_fakeredis_connector = MagicMock(spec=RedisConnector) mock_fakeredis_connector.ping.return_value = False mock_redis_connector_obj.side_effect = [ mock_redis_connector, mock_fakeredis_connector, ] with caplog.at_level(logging.INFO): with pytest.raises(RedisConnectorFactoryError): await redis_connector_factory() mock_redis_connector.ping.assert_called_once() mock_fakeredis_connector.ping.assert_called_once() assert caplog.records[0].levelno == logging.ERROR assert caplog.records[0].message == "Switching to use fakeredis."