File size: 1,696 Bytes
fd7800b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 | import unittest
from unittest.mock import AsyncMock, patch
from app.utils import cache
class RedisConfigurationTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.previous_url = cache.settings.redis_url
cache._redis_client = None
async def asyncTearDown(self):
cache.settings.redis_url = self.previous_url
cache._redis_client = None
async def test_missing_redis_does_not_create_cloud_database(self):
cache.settings.redis_url = ""
with (
patch("app.utils.cache._HAS_REDIS", True),
patch("app.utils.cache.aioredis.from_url") as from_url,
patch(
"app.utils.cache.provision_free_tier",
new=AsyncMock(return_value=None),
create=True,
) as provision,
):
result = await cache.get_redis()
self.assertIsNone(result)
from_url.assert_not_called()
provision.assert_not_awaited()
async def test_explicit_upstash_tls_url_is_used(self):
url = "rediss://default:placeholder@redis.example.test:6379/0"
cache.settings.redis_url = url
client = type("RedisClient", (), {"ping": AsyncMock(return_value=True)})()
with (
patch("app.utils.cache._HAS_REDIS", True),
patch("app.utils.cache.aioredis.from_url", return_value=client) as from_url,
):
result = await cache.get_redis()
self.assertIs(result, client)
from_url.assert_called_once()
self.assertEqual(from_url.call_args.args[0], url)
if __name__ == "__main__":
unittest.main()
|