65 lines
1.9 KiB
Python
65 lines
1.9 KiB
Python
import asyncio
|
|
import json
|
|
|
|
from api.core.Redis import get_redis_info, get_redis_info_async
|
|
from config.main import config
|
|
from database.connect import AsyncRedisPool, RedisPool
|
|
|
|
|
|
class FakeRedis:
|
|
def __init__(self, value=None, fail_write=False):
|
|
self.value = value
|
|
self.fail_write = fail_write
|
|
self.set_calls = []
|
|
|
|
def get(self, key):
|
|
return self.value
|
|
|
|
def set(self, key, value, **kwargs):
|
|
if self.fail_write:
|
|
raise RuntimeError('Redis unavailable')
|
|
self.set_calls.append((key, value, kwargs))
|
|
|
|
|
|
class FakeAsyncRedis(FakeRedis):
|
|
async def get(self, key):
|
|
return self.value
|
|
|
|
async def set(self, key, value, **kwargs):
|
|
super().set(key, value, **kwargs)
|
|
|
|
|
|
def test_shared_pool_keeps_resp2_compatibility():
|
|
assert RedisPool.connection_kwargs['protocol'] == 2
|
|
assert AsyncRedisPool.connection_kwargs['protocol'] == 2
|
|
|
|
|
|
def test_sync_cache_sets_value_and_ttl_in_one_command(monkeypatch):
|
|
monkeypatch.setattr(config, 'DEBUG_MODE', False)
|
|
redis = FakeRedis()
|
|
|
|
result = get_redis_info(redis, 'key', lambda: {'value': 1}, expire=60, without_gzip=True)
|
|
|
|
assert result == {'value': 1}
|
|
assert redis.set_calls == [('key', json.dumps(result), {'ex': 60})]
|
|
|
|
|
|
def test_sync_cache_write_failure_does_not_fail_request(monkeypatch):
|
|
monkeypatch.setattr(config, 'DEBUG_MODE', False)
|
|
redis = FakeRedis(fail_write=True)
|
|
|
|
assert get_redis_info(redis, 'key', lambda: {'value': 1}, without_gzip=True) == {'value': 1}
|
|
|
|
|
|
def test_async_cache_awaits_async_redis_io(monkeypatch):
|
|
monkeypatch.setattr(config, 'DEBUG_MODE', False)
|
|
redis = FakeAsyncRedis()
|
|
|
|
async def load():
|
|
return {'value': 1}
|
|
|
|
result = asyncio.run(get_redis_info_async(redis, 'key', load, expire=60, without_gzip=True))
|
|
|
|
assert result == {'value': 1}
|
|
assert redis.set_calls == [('key', json.dumps(result), {'ex': 60})]
|