40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from app.utils.redis_client import get_redis_client, reset_redis_clients_for_tests
|
|
|
|
|
|
class DomainCheckRedisClientTests(unittest.TestCase):
|
|
def tearDown(self) -> None:
|
|
reset_redis_clients_for_tests()
|
|
|
|
@patch("app.utils.redis_client.redis.Redis")
|
|
@patch("app.utils.redis_client.redis.BlockingConnectionPool")
|
|
def test_standard_client_is_cached_per_process(self, mock_pool, mock_redis) -> None:
|
|
client = object()
|
|
mock_redis.return_value = client
|
|
|
|
first = get_redis_client()
|
|
second = get_redis_client()
|
|
|
|
self.assertIs(first, client)
|
|
self.assertIs(second, client)
|
|
mock_pool.assert_called_once()
|
|
mock_redis.assert_called_once()
|
|
|
|
@patch("app.utils.redis_client.redis.Redis")
|
|
@patch("app.utils.redis_client.redis.BlockingConnectionPool")
|
|
def test_pubsub_role_uses_separate_cached_client(self, mock_pool, mock_redis) -> None:
|
|
mock_redis.side_effect = [object(), object()]
|
|
|
|
standard_client = get_redis_client()
|
|
pubsub_client = get_redis_client(role="pubsub")
|
|
|
|
self.assertIsNot(standard_client, pubsub_client)
|
|
self.assertEqual(2, mock_pool.call_count)
|
|
self.assertEqual(2, mock_redis.call_count)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|