forked from mirror/ledisdb
122 lines
2.7 KiB
Python
122 lines
2.7 KiB
Python
# coding: utf-8
|
|
# Test Cases for other commands
|
|
|
|
import unittest
|
|
import sys
|
|
sys.path.append('..')
|
|
|
|
import ledis
|
|
from ledis._compat import b
|
|
from ledis import ResponseError
|
|
|
|
l = ledis.Ledis(port=6380)
|
|
dbs = ["leveldb", "rocksdb", "goleveldb", "hyperleveldb", "lmdb", "boltdb"]
|
|
|
|
class TestOtherCommands(unittest.TestCase):
|
|
def setUp(self):
|
|
pass
|
|
|
|
def tearDown(self):
|
|
l.flushdb()
|
|
|
|
# server information
|
|
def test_echo(self):
|
|
assert l.echo('foo bar') == b('foo bar')
|
|
|
|
def test_ping(self):
|
|
assert l.ping()
|
|
|
|
def test_select(self):
|
|
assert l.select('1')
|
|
assert l.select('15')
|
|
self.assertRaises(ResponseError, lambda: l.select('16'))
|
|
|
|
|
|
def test_info(self):
|
|
info1 = l.info()
|
|
assert info1.get("db_name") in dbs
|
|
info2 = l.info(section="server")
|
|
assert info2.get("os") in ["linux", "darwin"]
|
|
|
|
def test_flushdb(self):
|
|
l.set("a", 1)
|
|
assert l.flushdb() == "OK"
|
|
assert l.get("a") is None
|
|
|
|
def test_flushall(self):
|
|
l.select(1)
|
|
l.set("a", 1)
|
|
assert l.get("a") == b("1")
|
|
|
|
l.select(10)
|
|
l.set("a", 1)
|
|
assert l.get("a") == b("1")
|
|
|
|
assert l.flushall() == "OK"
|
|
|
|
assert l.get("a") is None
|
|
l.select(1)
|
|
assert l.get("a") is None
|
|
|
|
|
|
# test *scan commands
|
|
|
|
def check_keys(self, scan_type):
|
|
d = {
|
|
"scan": l.scan,
|
|
"sscan": l.sscan,
|
|
"lscan": l.lscan,
|
|
"hscan": l.hscan,
|
|
"zscan": l.zscan,
|
|
"bscan": l.bscan
|
|
}
|
|
|
|
key, keys = d[scan_type]()
|
|
assert key == ""
|
|
assert set(keys) == set([b("a"), b("b"), b("c")])
|
|
|
|
_, keys = d[scan_type](match="a")
|
|
assert set(keys) == set([b("a")])
|
|
|
|
_, keys = d[scan_type](key="a")
|
|
assert set(keys) == set([b("b"), b("c")])
|
|
|
|
|
|
def test_scan(self):
|
|
d = {"a":1, "b":2, "c": 3}
|
|
l.mset(d)
|
|
self.check_keys("scan")
|
|
|
|
|
|
def test_lscan(self):
|
|
l.rpush("a", 1)
|
|
l.rpush("b", 1)
|
|
l.rpush("c", 1)
|
|
self.check_keys("lscan")
|
|
|
|
|
|
def test_hscan(self):
|
|
l.hset("a", "hello", "world")
|
|
l.hset("b", "hello", "world")
|
|
l.hset("c", "hello", "world")
|
|
self.check_keys("hscan")
|
|
|
|
def test_sscan(self):
|
|
l.sadd("a", 1)
|
|
l.sadd("b", 2)
|
|
l.sadd("c", 3)
|
|
self.check_keys("sscan")
|
|
|
|
def test_zscan(self):
|
|
l.zadd("a", 1, "a")
|
|
l.zadd("b", 1, "a")
|
|
l.zadd("c", 1, "a")
|
|
self.check_keys("zscan")
|
|
|
|
def test_bscan(self):
|
|
l.bsetbit("a", 1, 1)
|
|
l.bsetbit("b", 1, 1)
|
|
l.bsetbit("c", 1, 1)
|
|
self.check_keys("bscan")
|
|
|