|
|
@ -1,5 +1,5 @@
|
|
|
|
#!/usr/bin/python
|
|
|
|
#!/usr/bin/python3
|
|
|
|
import urllib2, unittest, json, hashlib
|
|
|
|
import urllib.request, urllib.error, urllib.parse, unittest, json, hashlib
|
|
|
|
from functools import wraps
|
|
|
|
from functools import wraps
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
import msgpack
|
|
|
|
import msgpack
|
|
|
@ -16,15 +16,15 @@ class TestWebdis(unittest.TestCase):
|
|
|
|
return 'http://%s:%d/%s' % (host, port, url)
|
|
|
|
return 'http://%s:%d/%s' % (host, port, url)
|
|
|
|
|
|
|
|
|
|
|
|
def query(self, url, data = None, headers={}):
|
|
|
|
def query(self, url, data = None, headers={}):
|
|
|
|
r = urllib2.Request(self.wrap(url), data, headers)
|
|
|
|
r = urllib.request.Request(self.wrap(url), data, headers)
|
|
|
|
return urllib2.urlopen(r)
|
|
|
|
return urllib.request.urlopen(r)
|
|
|
|
|
|
|
|
|
|
|
|
class TestBasics(TestWebdis):
|
|
|
|
class TestBasics(TestWebdis):
|
|
|
|
|
|
|
|
|
|
|
|
def test_crossdomain(self):
|
|
|
|
def test_crossdomain(self):
|
|
|
|
f = self.query('crossdomain.xml')
|
|
|
|
f = self.query('crossdomain.xml')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/xml')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/xml')
|
|
|
|
self.assertTrue("allow-access-from domain" in f.read())
|
|
|
|
self.assertTrue(b"allow-access-from domain" in f.read())
|
|
|
|
|
|
|
|
|
|
|
|
def test_options(self):
|
|
|
|
def test_options(self):
|
|
|
|
pass
|
|
|
|
pass
|
|
|
@ -42,25 +42,25 @@ class TestJSON(TestWebdis):
|
|
|
|
"success type (+OK)"
|
|
|
|
"success type (+OK)"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
f = self.query('SET/hello/world')
|
|
|
|
f = self.query('SET/hello/world')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.headers.getheader('ETag') == '"0db1124cf79ffeb80aff6d199d5822f8"')
|
|
|
|
self.assertTrue(f.getheader('ETag') == '"0db1124cf79ffeb80aff6d199d5822f8"')
|
|
|
|
self.assertTrue(f.read() == '{"SET":[true,"OK"]}')
|
|
|
|
self.assertTrue(f.read() == b'{"SET":[true,"OK"]}')
|
|
|
|
|
|
|
|
|
|
|
|
def test_get(self):
|
|
|
|
def test_get(self):
|
|
|
|
"string type"
|
|
|
|
"string type"
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
f = self.query('GET/hello')
|
|
|
|
f = self.query('GET/hello')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.headers.getheader('ETag') == '"8cf38afc245b7a6a88696566483d1390"')
|
|
|
|
self.assertTrue(f.getheader('ETag') == '"8cf38afc245b7a6a88696566483d1390"')
|
|
|
|
self.assertTrue(f.read() == '{"GET":"world"}')
|
|
|
|
self.assertTrue(f.read() == b'{"GET":"world"}')
|
|
|
|
|
|
|
|
|
|
|
|
def test_incr(self):
|
|
|
|
def test_incr(self):
|
|
|
|
"integer type"
|
|
|
|
"integer type"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
f = self.query('INCR/hello')
|
|
|
|
f = self.query('INCR/hello')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.headers.getheader('ETag') == '"500e9bcdcbb1e98f25c1fbb880a96c99"')
|
|
|
|
self.assertTrue(f.getheader('ETag') == '"500e9bcdcbb1e98f25c1fbb880a96c99"')
|
|
|
|
self.assertTrue(f.read() == '{"INCR":1}')
|
|
|
|
self.assertTrue(f.read() == b'{"INCR":1}')
|
|
|
|
|
|
|
|
|
|
|
|
def test_list(self):
|
|
|
|
def test_list(self):
|
|
|
|
"list type"
|
|
|
|
"list type"
|
|
|
@ -68,16 +68,16 @@ class TestJSON(TestWebdis):
|
|
|
|
self.query('RPUSH/hello/abc')
|
|
|
|
self.query('RPUSH/hello/abc')
|
|
|
|
self.query('RPUSH/hello/def')
|
|
|
|
self.query('RPUSH/hello/def')
|
|
|
|
f = self.query('LRANGE/hello/0/-1')
|
|
|
|
f = self.query('LRANGE/hello/0/-1')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.headers.getheader('ETag') == '"622e51f547a480bef7cf5452fb7782db"')
|
|
|
|
self.assertTrue(f.getheader('ETag') == '"622e51f547a480bef7cf5452fb7782db"')
|
|
|
|
self.assertTrue(f.read() == '{"LRANGE":["abc","def"]}')
|
|
|
|
self.assertTrue(f.read() == b'{"LRANGE":["abc","def"]}')
|
|
|
|
|
|
|
|
|
|
|
|
def test_error(self):
|
|
|
|
def test_error(self):
|
|
|
|
"error return type"
|
|
|
|
"error return type"
|
|
|
|
f = self.query('UNKNOWN/COMMAND')
|
|
|
|
f = self.query('UNKNOWN/COMMAND')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/json')
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
obj = json.loads(f.read())
|
|
|
|
obj = json.loads(f.read().decode('utf-8'))
|
|
|
|
except:
|
|
|
|
except:
|
|
|
|
self.assertTrue(False)
|
|
|
|
self.assertTrue(False)
|
|
|
|
return
|
|
|
|
return
|
|
|
@ -86,7 +86,7 @@ class TestJSON(TestWebdis):
|
|
|
|
self.assertTrue('UNKNOWN' in obj)
|
|
|
|
self.assertTrue('UNKNOWN' in obj)
|
|
|
|
self.assertTrue(isinstance(obj['UNKNOWN'], list))
|
|
|
|
self.assertTrue(isinstance(obj['UNKNOWN'], list))
|
|
|
|
self.assertTrue(obj['UNKNOWN'][0] == False)
|
|
|
|
self.assertTrue(obj['UNKNOWN'][0] == False)
|
|
|
|
self.assertTrue(isinstance(obj['UNKNOWN'][1], unicode))
|
|
|
|
self.assertTrue(isinstance(obj['UNKNOWN'][1], str))
|
|
|
|
|
|
|
|
|
|
|
|
class TestCustom(TestWebdis):
|
|
|
|
class TestCustom(TestWebdis):
|
|
|
|
def test_list(self):
|
|
|
|
def test_list(self):
|
|
|
@ -94,16 +94,16 @@ class TestCustom(TestWebdis):
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('RPUSH/hello/a/b/c')
|
|
|
|
self.query('RPUSH/hello/a/b/c')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.txt')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.txt')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'text/plain')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'text/plain')
|
|
|
|
self.assertTrue(f.read() == "abc")
|
|
|
|
self.assertTrue(f.read() == b"abc")
|
|
|
|
|
|
|
|
|
|
|
|
def test_separator(self):
|
|
|
|
def test_separator(self):
|
|
|
|
"Separator in list responses with custom format"
|
|
|
|
"Separator in list responses with custom format"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('RPUSH/hello/a/b/c')
|
|
|
|
self.query('RPUSH/hello/a/b/c')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.txt?sep=--')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.txt?sep=--')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'text/plain')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'text/plain')
|
|
|
|
self.assertTrue(f.read() == "a--b--c")
|
|
|
|
self.assertTrue(f.read() == b"a--b--c")
|
|
|
|
|
|
|
|
|
|
|
|
class TestRaw(TestWebdis):
|
|
|
|
class TestRaw(TestWebdis):
|
|
|
|
|
|
|
|
|
|
|
@ -111,20 +111,20 @@ class TestRaw(TestWebdis):
|
|
|
|
"success type (+OK)"
|
|
|
|
"success type (+OK)"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
f = self.query('SET/hello/world.raw')
|
|
|
|
f = self.query('SET/hello/world.raw')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'binary/octet-stream')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'binary/octet-stream')
|
|
|
|
self.assertTrue(f.read() == "+OK\r\n")
|
|
|
|
self.assertTrue(f.read() == b"+OK\r\n")
|
|
|
|
|
|
|
|
|
|
|
|
def test_get(self):
|
|
|
|
def test_get(self):
|
|
|
|
"string type"
|
|
|
|
"string type"
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
f = self.query('GET/hello.raw')
|
|
|
|
f = self.query('GET/hello.raw')
|
|
|
|
self.assertTrue(f.read() == '$5\r\nworld\r\n')
|
|
|
|
self.assertTrue(f.read() == b'$5\r\nworld\r\n')
|
|
|
|
|
|
|
|
|
|
|
|
def test_incr(self):
|
|
|
|
def test_incr(self):
|
|
|
|
"integer type"
|
|
|
|
"integer type"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
f = self.query('INCR/hello.raw')
|
|
|
|
f = self.query('INCR/hello.raw')
|
|
|
|
self.assertTrue(f.read() == ':1\r\n')
|
|
|
|
self.assertTrue(f.read() == b':1\r\n')
|
|
|
|
|
|
|
|
|
|
|
|
def test_list(self):
|
|
|
|
def test_list(self):
|
|
|
|
"list type"
|
|
|
|
"list type"
|
|
|
@ -132,12 +132,12 @@ class TestRaw(TestWebdis):
|
|
|
|
self.query('RPUSH/hello/abc')
|
|
|
|
self.query('RPUSH/hello/abc')
|
|
|
|
self.query('RPUSH/hello/def')
|
|
|
|
self.query('RPUSH/hello/def')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.raw')
|
|
|
|
f = self.query('LRANGE/hello/0/-1.raw')
|
|
|
|
self.assertTrue(f.read() == "*2\r\n$3\r\nabc\r\n$3\r\ndef\r\n")
|
|
|
|
self.assertTrue(f.read() == b"*2\r\n$3\r\nabc\r\n$3\r\ndef\r\n")
|
|
|
|
|
|
|
|
|
|
|
|
def test_error(self):
|
|
|
|
def test_error(self):
|
|
|
|
"error return type"
|
|
|
|
"error return type"
|
|
|
|
f = self.query('UNKNOWN/COMMAND.raw')
|
|
|
|
f = self.query('UNKNOWN/COMMAND.raw')
|
|
|
|
self.assertTrue(f.read().startswith("-ERR "))
|
|
|
|
self.assertTrue(f.read().startswith(b"-ERR "))
|
|
|
|
|
|
|
|
|
|
|
|
def need_msgpack(fn):
|
|
|
|
def need_msgpack(fn):
|
|
|
|
def wrapper(self):
|
|
|
|
def wrapper(self):
|
|
|
@ -152,7 +152,7 @@ class TestMsgPack(TestWebdis):
|
|
|
|
"success type (+OK)"
|
|
|
|
"success type (+OK)"
|
|
|
|
self.query('DEL/hello')
|
|
|
|
self.query('DEL/hello')
|
|
|
|
f = self.query('SET/hello/world.msg')
|
|
|
|
f = self.query('SET/hello/world.msg')
|
|
|
|
self.assertTrue(f.headers.getheader('Content-Type') == 'application/x-msgpack')
|
|
|
|
self.assertTrue(f.getheader('Content-Type') == 'application/x-msgpack')
|
|
|
|
obj = msgpack.loads(f.read())
|
|
|
|
obj = msgpack.loads(f.read())
|
|
|
|
self.assertTrue(obj == {'SET': (True, 'OK')})
|
|
|
|
self.assertTrue(obj == {'SET': (True, 'OK')})
|
|
|
|
|
|
|
|
|
|
|
@ -197,19 +197,19 @@ class TestETag(TestWebdis):
|
|
|
|
|
|
|
|
|
|
|
|
def test_etag_match(self):
|
|
|
|
def test_etag_match(self):
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
h = hashlib.md5("world").hexdigest() # match Etag
|
|
|
|
h = hashlib.md5("world".encode()).hexdigest() # match Etag
|
|
|
|
try:
|
|
|
|
try:
|
|
|
|
f = self.query('GET/hello.txt', None, {'If-None-Match': '"'+ h +'"'})
|
|
|
|
f = self.query('GET/hello.txt', None, {'If-None-Match': '"'+ h +'"'})
|
|
|
|
except urllib2.HTTPError as e:
|
|
|
|
except urllib.error.HTTPError as e:
|
|
|
|
self.assertTrue(e.code == 304)
|
|
|
|
self.assertTrue(e.code == 304)
|
|
|
|
return
|
|
|
|
return
|
|
|
|
self.assertTrue(False) # we should have received a 304.
|
|
|
|
self.assertTrue(False) # we should have received a 304.
|
|
|
|
|
|
|
|
|
|
|
|
def test_etag_fail(self):
|
|
|
|
def test_etag_fail(self):
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
self.query('SET/hello/world')
|
|
|
|
h = hashlib.md5("nonsense").hexdigest() # non-matching Etag
|
|
|
|
h = hashlib.md5("nonsense".encode()).hexdigest() # non-matching Etag
|
|
|
|
f = self.query('GET/hello.txt', None, {'If-None-Match': '"'+ h +'"'})
|
|
|
|
f = self.query('GET/hello.txt', None, {'If-None-Match': '"'+ h +'"'})
|
|
|
|
self.assertTrue(f.read() == 'world')
|
|
|
|
self.assertTrue(f.read() == b'world')
|
|
|
|
|
|
|
|
|
|
|
|
class TestDbSwitch(TestWebdis):
|
|
|
|
class TestDbSwitch(TestWebdis):
|
|
|
|
def test_db(self):
|
|
|
|
def test_db(self):
|
|
|
@ -217,11 +217,11 @@ class TestDbSwitch(TestWebdis):
|
|
|
|
self.query('0/SET/key/val0')
|
|
|
|
self.query('0/SET/key/val0')
|
|
|
|
self.query('1/SET/key/val1')
|
|
|
|
self.query('1/SET/key/val1')
|
|
|
|
f = self.query('0/GET/key.txt')
|
|
|
|
f = self.query('0/GET/key.txt')
|
|
|
|
self.assertTrue(f.read() == "val0")
|
|
|
|
self.assertTrue(f.read() == b"val0")
|
|
|
|
f = self.query('1/GET/key.txt')
|
|
|
|
f = self.query('1/GET/key.txt')
|
|
|
|
self.assertTrue(f.read() == "val1")
|
|
|
|
self.assertTrue(f.read() == b"val1")
|
|
|
|
f = self.query('GET/key.txt')
|
|
|
|
f = self.query('GET/key.txt')
|
|
|
|
self.assertTrue(f.read() == "val0")
|
|
|
|
self.assertTrue(f.read() == b"val0")
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
if __name__ == '__main__':
|
|
|
|
unittest.main()
|
|
|
|
unittest.main()
|
|
|
|