Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_set_config_no_default(self):
with pytest.raises(ValueError):
caches.set_config(
{
"no_default": {
"cache": "aiocache.RedisCache",
"endpoint": "127.0.0.10",
"port": 6378,
"serializer": {"class": "aiocache.serializers.PickleSerializer"},
"plugins": [
{"class": "aiocache.plugins.HitMissRatioPlugin"},
{"class": "aiocache.plugins.TimingPlugin"},
],
def test_ensure_plugins_order(self):
caches.set_config(
{
"default": {
"cache": "aiocache.RedisCache",
"plugins": [
{"class": "aiocache.plugins.HitMissRatioPlugin"},
{"class": "aiocache.plugins.TimingPlugin"},
],
}
}
)
cache = caches.get("default")
assert isinstance(cache.plugins[0], HitMissRatioPlugin)
cache = caches.create("default")
assert isinstance(cache.plugins[0], HitMissRatioPlugin)
def test_multiple_caches(self):
caches.set_config(
{
"default": {
"cache": "aiocache.RedisCache",
"endpoint": "127.0.0.10",
"port": 6378,
"serializer": {"class": "aiocache.serializers.PickleSerializer"},
"plugins": [
{"class": "aiocache.plugins.HitMissRatioPlugin"},
{"class": "aiocache.plugins.TimingPlugin"},
],
},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
default = caches.get("default")
def test_set_config_removes_existing_caches(self):
caches.set_config(
{
"default": {"cache": "aiocache.SimpleMemoryCache"},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
caches.get("default")
caches.get("alt")
assert len(caches._caches) == 2
caches.set_config(
{
"default": {"cache": "aiocache.SimpleMemoryCache"},
"alt": {"cache": "aiocache.SimpleMemoryCache"},
}
)
assert caches._caches == {}
def test_set_empty_config(self):
with pytest.raises(ValueError):
caches.set_config({})
import asyncio
from aiocache import caches, Cache
from aiocache.serializers import StringSerializer, PickleSerializer
caches.set_config({
'default': {
'cache': "aiocache.SimpleMemoryCache",
'serializer': {
'class': "aiocache.serializers.StringSerializer"
}
},
'redis_alt': {
'cache': "aiocache.RedisCache",
'endpoint': "127.0.0.1",
'port': 6379,
'timeout': 1,
'serializer': {
'class': "aiocache.serializers.PickleSerializer"
},
'plugins': [
{'class': "aiocache.plugins.HitMissRatioPlugin"},
'port': os.getenv('CACHE_PORT'),
'password': os.getenv('CACHE_PASSWORD'),
'db': os.getenv('CACHE_DB'),
'namespace': 'hero_' + os.getenv('NAMESPACE'),
'pool_min_size': 1,
'pool_max_size': 10,
'serializer': {
'class': 'aiocache.serializers.PickleSerializer'
}
}
}
else:
raise ConfigurationError("The configuration uses an unsupported cache backend: "
"{}".format(os.getenv('CACHE_TYPE')))
aiocache.caches.set_config(_cache_config)
def set_config(cls, config):
logger.warning("Deprecated, please use 'aiocache.caches.get_config'")
return caches.set_config(config)