| 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() | |