Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_redlock_release(self, memory):
SimpleMemoryBackend._cache.get.return_value = "lock"
assert await memory._redlock_release(pytest.KEY, "lock") == 1
SimpleMemoryBackend._cache.get.assert_called_with(pytest.KEY)
SimpleMemoryBackend._cache.pop.assert_called_with(pytest.KEY)
async def test_clear_no_namespace(self, memory):
SimpleMemoryBackend._handlers = "asdad"
SimpleMemoryBackend._cache = "asdad"
await memory._clear()
SimpleMemoryBackend._handlers = {}
SimpleMemoryBackend._cache = {}
async def test_increment(self, memory):
await memory._increment(pytest.KEY, 2)
SimpleMemoryBackend._cache.__contains__.assert_called_with(pytest.KEY)
SimpleMemoryBackend._cache.__setitem__.assert_called_with(pytest.KEY, 2)
async def test_multi_set(self, memory):
await memory._multi_set([(pytest.KEY, "value"), (pytest.KEY_1, "random")])
SimpleMemoryBackend._cache.__setitem__.assert_any_call(pytest.KEY, "value")
SimpleMemoryBackend._cache.__setitem__.assert_any_call(pytest.KEY_1, "random")
async def test_clear_namespace(self, memory):
SimpleMemoryBackend._cache.__iter__.return_value = iter(["nma", "nmb", "no"])
await memory._clear("nm")
assert SimpleMemoryBackend._cache.pop.call_count == 2
SimpleMemoryBackend._cache.pop.assert_any_call("nma", None)
SimpleMemoryBackend._cache.pop.assert_any_call("nmb", None)
async def test_increment_missing(self, memory):
SimpleMemoryBackend._cache.__contains__.return_value = True
SimpleMemoryBackend._cache.__getitem__.return_value = 2
await memory._increment(pytest.KEY, 2)
SimpleMemoryBackend._cache.__getitem__.assert_called_with(pytest.KEY)
SimpleMemoryBackend._cache.__setitem__.assert_called_with(pytest.KEY, 4)
async def test_get_avg_min_max(self, memory_cache, data, ratio):
keys = ["a", "b", "c", "d", "e", "f"]
memory_cache.plugins = [TimingPlugin()]
SimpleMemoryBackend._cache = data
for key in keys:
await memory_cache.get(key)
assert "get_max" in memory_cache.profiling
assert "get_min" in memory_cache.profiling
assert "get_total" in memory_cache.profiling
assert "get_avg" in memory_cache.profiling
async def _clear(self, namespace=None, _conn=None):
if namespace:
for key in list(SimpleMemoryBackend._cache):
if key.startswith(namespace):
self.__delete(key)
else:
SimpleMemoryBackend._cache = {}
SimpleMemoryBackend._handlers = {}
return True
async def _redlock_release(self, key, value):
if SimpleMemoryBackend._cache.get(key) == value:
SimpleMemoryBackend._cache.pop(key)
return 1
return 0
async def _multi_get(self, keys, encoding="utf-8", _conn=None):
return [SimpleMemoryBackend._cache.get(key) for key in keys]