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_redis_getter_setter(tmpdir):
""":type tmpdir: py.path.local"""
from cache_requests import Session
from redislite import StrictRedis
# LOCAL SETUP
# ------------------------------------------------------------------------
request = Session()
test_db = tmpdir.join('test_redis.db').strpath
test_connection = request.cache.connection
alt_db = tmpdir.join('test_redis_getter_setter.db').strpath
alt_connection = StrictRedis(dbfilename=alt_db)
# TEST SETUP
# ------------------------------------------------------------------------
assert test_connection.db == test_db
assert alt_connection.db == alt_db
assert test_db != alt_db
# TEST SESSION CONNECTION IDENTITY WITH METHOD's REDIS HANDLE
# ------------------------------------------------------------------------
assert request.get.redis is request.cache.connection
assert request.cache.connection.db == test_db
request.post.redis = alt_connection
assert request.post.redis is request.patch.redis
def amazing_function():
from cache_requests import config
from redislite import StrictRedis
from cache_requests import Memoize
def _side_effect(*args, **kwargs):
return len(args), len(kwargs)
_amazing_function = MagicMock(spec=_side_effect, side_effect=_side_effect)
connection = StrictRedis(dbfilename=config.dbfilename)
return Memoize(_amazing_function, ex=1, connection=connection)
def test_redis_log_tail_empty_log(self):
r = redislite.StrictRedis()
with open(r.logfile, 'w'):
pass
lines = r.redis_log_tail()
self.assertEqual(lines, [])
def test_redis_log_tail(self):
r = redislite.StrictRedis()
lines = r.redis_log_tail(2)
self.assertIsInstance(lines, list)
self.assertEqual(len(lines), 2)
def test_redis_log_small_chunks(self):
r = redislite.StrictRedis()
lines = r.redis_log_tail(4, width=20)
self.assertIsInstance(lines, list)
self.assertEqual(len(lines), 4)
def test_redis_log_tail_no_log(self):
r = redislite.StrictRedis()
if os.path.exists(r.logfile):
os.remove(r.logfile)
lines = r.redis_log_tail()
self.assertEqual(lines, [])
def test_redis_log_attribute(self):
r = redislite.StrictRedis()
self.assertIn('The server is now ready to accept connections', r.redis_log)
def build_redis(port):
if port is None:
client = redislite.StrictRedis()
else:
client = redis.StrictRedis(port=int(port))
redpipe.reset()
redpipe.connect_redis(client)
return client
"""
from __future__ import absolute_import
from functools import partial
from os import environ, path
from tempfile import gettempdir
from redislite import StrictRedis
temp_file = partial(path.join, gettempdir())
__all__ = ['ex', 'dbfilename', 'connection']
ex = environ.get('REDIS_EX', 60 * 60) # 1 hour
dbfilename = environ.get('REDIS_DBFILENAME', temp_file('cache_requests.redislite'))
connection = environ.get('REDIS_CONNECTION') or partial(StrictRedis, dbfilename=dbfilename)