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_redislite_Redis_multiple_connections(self):
# Generate a new redis server
r = redislite.Redis()
self._log_redis_pid(r)
# Pass the first server's db to get a second connection
s = redislite.Redis(r.db)
self._log_redis_pid(s)
r.set('key', 'value')
result = s.get('key').decode(encoding='UTF-8')
self.assertEqual(result, 'value')
# Both objects should be using the same redis process.
self.assertEqual(r.pid, s.pid)
def test_redislite_Redis_cleanup(self):
r = redislite.Redis()
self._log_redis_pid(r)
r._cleanup()
self.assertIsNone(r.socket_file)
def test_auth(self):
redis_server = redislite.Redis(
serverconfig={
'requirepass': 'test',
'port': self.redis_test_port+1
},
password='test'
)
# This shouldn't generate an exception
try:
redis_client = redis.Redis(host='127.0.0.1', port=self.redis_test_port+1, password='test')
uredis_client = uredis.Redis(host='127.0.0.1', port=self.redis_test_port+1, password='test')
finally:
redis_server.shutdown()
def test_redislite_Redis_create_redis_directory_tree(self):
r = redislite.Redis()
self._log_redis_pid(r)
r._create_redis_directory_tree()
self.assertTrue(r.redis_dir)
self.assertTrue(os.path.exists(r.redis_dir))
self.assertTrue(os.path.exists(r.pidfile))
self.assertTrue(os.path.exists(r.socket_file))
r._cleanup()
def test_redislite_db_file_cwd_kw(self):
test_db = 'test_unit_redis.db'
if os.path.exists(test_db):
os.remove(test_db)
r = redislite.Redis(dbfilename=test_db)
self._log_redis_pid(r)
r.set('key', 'value')
r.save()
self.assertTrue(os.path.exists(test_db))
os.remove(test_db)
def test_redislite_redis_custom_socket_file(self):
"""
Test creating a redis instance with a specified socket filename
:return:
"""
socket_file_name = '/tmp/test.socket'
r = redislite.Redis(unix_socket_path=socket_file_name)
self._log_redis_pid(r)
self.assertEqual(r.socket_file, socket_file_name)
print(os.listdir('.'))
mode = os.stat(socket_file_name).st_mode
isSocket = stat.S_ISSOCK(mode)
self.assertTrue(isSocket)
r._cleanup()
def test_redislite_Redis(self):
r = redislite.Redis()
self._log_redis_pid(r)
r.set('key', 'value')
result = r.get('key').decode(encoding='UTF-8')
self.assertEqual(result, 'value')
def setUp(self):
self.redis_server = redislite.Redis(serverconfig={'port': self.redis_test_port})
self.uredis_client = uredis.Redis(host='127.0.0.1', port=self.redis_test_port)
def test_redislite_Redis_with_db_file_keyword(self):
temp_dir = tempfile.mkdtemp()
filename = os.path.join(temp_dir, 'redis.db')
self.assertFalse(os.path.exists(filename))
r = redislite.Redis(dbfilename=filename)
self._log_redis_pid(r)
r.set('key', 'value')
result = r.get('key').decode(encoding='UTF-8')
self.assertEqual(result, 'value')
r.save()
r._cleanup()
self.assertTrue(os.path.exists(filename))
shutil.rmtree(temp_dir)
def _ensure_redis(redis: Optional[Redis]):
if redis:
return redis
try:
import redislite
return redislite.Redis()
except ImportError:
raise ValueError(
"Redis instance not given and redislite not importable. Run\n"
"pip install redislite"