#!/usr/bin/python
-import urllib2, unittest, json, hashlib
+import urllib.request, urllib.error, urllib.parse, unittest, json, hashlib
from functools import wraps
try:
import msgpack
return 'http://%s:%d/%s' % (host, port, url)
def query(self, url, data = None, headers={}):
- r = urllib2.Request(self.wrap(url), data, headers)
- return urllib2.urlopen(r)
+ r = urllib.request.Request(self.wrap(url), data, headers)
+ return urllib.request.urlopen(r)
class TestBasics(TestWebdis):
def test_crossdomain(self):
f = self.query('crossdomain.xml')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/xml')
- self.assertTrue("allow-access-from domain" in f.read())
+ self.assertTrue(f.getheader('Content-Type') == 'application/xml')
+ self.assertTrue(b"allow-access-from domain" in f.read())
def test_options(self):
pass
"success type (+OK)"
self.query('DEL/hello')
f = self.query('SET/hello/world')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
- self.assertTrue(f.headers.getheader('ETag') == '"0db1124cf79ffeb80aff6d199d5822f8"')
- self.assertTrue(f.read() == '{"SET":[true,"OK"]}')
+ self.assertTrue(f.getheader('Content-Type') == 'application/json')
+ self.assertTrue(f.getheader('ETag') == '"0db1124cf79ffeb80aff6d199d5822f8"')
+ self.assertTrue(f.read() == b'{"SET":[true,"OK"]}')
def test_get(self):
"string type"
self.query('SET/hello/world')
f = self.query('GET/hello')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
- self.assertTrue(f.headers.getheader('ETag') == '"8cf38afc245b7a6a88696566483d1390"')
- self.assertTrue(f.read() == '{"GET":"world"}')
+ self.assertTrue(f.getheader('Content-Type') == 'application/json')
+ self.assertTrue(f.getheader('ETag') == '"8cf38afc245b7a6a88696566483d1390"')
+ self.assertTrue(f.read() == b'{"GET":"world"}')
def test_incr(self):
"integer type"
self.query('DEL/hello')
f = self.query('INCR/hello')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
- self.assertTrue(f.headers.getheader('ETag') == '"500e9bcdcbb1e98f25c1fbb880a96c99"')
- self.assertTrue(f.read() == '{"INCR":1}')
+ self.assertTrue(f.getheader('Content-Type') == 'application/json')
+ self.assertTrue(f.getheader('ETag') == '"500e9bcdcbb1e98f25c1fbb880a96c99"')
+ self.assertTrue(f.read() == b'{"INCR":1}')
def test_list(self):
"list type"
self.query('RPUSH/hello/abc')
self.query('RPUSH/hello/def')
f = self.query('LRANGE/hello/0/-1')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
- self.assertTrue(f.headers.getheader('ETag') == '"622e51f547a480bef7cf5452fb7782db"')
- self.assertTrue(f.read() == '{"LRANGE":["abc","def"]}')
+ self.assertTrue(f.getheader('Content-Type') == 'application/json')
+ self.assertTrue(f.getheader('ETag') == '"622e51f547a480bef7cf5452fb7782db"')
+ self.assertTrue(f.read() == b'{"LRANGE":["abc","def"]}')
def test_error(self):
"error return type"
f = self.query('UNKNOWN/COMMAND')
- self.assertTrue(f.headers.getheader('Content-Type') == 'application/json')
+ self.assertTrue(f.getheader('Content-Type') == 'application/json')
try:
obj = json.loads(f.read())
except:
self.assertTrue('UNKNOWN' in obj)
self.assertTrue(isinstance(obj['UNKNOWN'], list))
self.assertTrue(obj['UNKNOWN'][0] == False)
- self.assertTrue(isinstance(obj['UNKNOWN'][1], unicode))
+ self.assertTrue(isinstance(obj['UNKNOWN'][1], str))
class TestCustom(TestWebdis):
def test_list(self):
self.query('DEL/hello')
self.query('RPUSH/hello/a/b/c')
f = self.query('LRANGE/hello/0/-1.txt')
- self.assertTrue(f.headers.getheader('Content-Type') == 'text/plain')
- self.assertTrue(f.read() == "abc")
+ self.assertTrue(f.getheader('Content-Type') == 'text/plain')
+ self.assertTrue(f.read() == b"abc")
def test_separator(self):
"Separator in list responses with custom format"
self.query('DEL/hello')
self.query('RPUSH/hello/a/b/c')
f = self.query('LRANGE/hello/0/-1.txt?sep=--')
- self.assertTrue(f.headers.getheader('Content-Type') == 'text/plain')
- self.assertTrue(f.read() == "a--b--c")
+ self.assertTrue(f.getheader('Content-Type') == 'text/plain')
+ self.assertTrue(f.read() == b"a--b--c")
class TestRaw(TestWebdis):
"success type (+OK)"
self.query('DEL/hello')
f = self.query('SET/hello/world.raw')
- self.assertTrue(f.headers.getheader('Content-Type') == 'binary/octet-stream')
- self.assertTrue(f.read() == "+OK\r\n")
+ self.assertTrue(f.getheader('Content-Type') == 'binary/octet-stream')
+ self.assertTrue(f.read() == b"+OK\r\n")
def test_get(self):
"string type"
self.query('SET/hello/world')
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):
"integer type"
self.query('DEL/hello')
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):
"list type"
self.query('RPUSH/hello/abc')
self.query('RPUSH/hello/def')
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):
"error return type"
f = self.query('UNKNOWN/COMMAND.raw')
- self.assertTrue(f.read().startswith("-ERR "))
+ self.assertTrue(f.read().startswith(b"-ERR "))
def need_msgpack(fn):
def wrapper(self):
"success type (+OK)"
self.query('DEL/hello')
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())
self.assertTrue(obj == {'SET': [True, 'OK']})
def test_etag_match(self):
self.query('SET/hello/world')
- h = hashlib.md5("world").hexdigest() # match Etag
+ h = hashlib.md5("world".encode()).hexdigest() # match Etag
try:
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)
return
self.assertTrue(False) # we should have received a 304.
def test_etag_fail(self):
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 +'"'})
- self.assertTrue(f.read() == 'world')
+ self.assertTrue(f.read() == b'world')
class TestDbSwitch(TestWebdis):
def test_db(self):
self.query('0/SET/key/val0')
self.query('1/SET/key/val1')
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')
- self.assertTrue(f.read() == "val1")
+ self.assertTrue(f.read() == b"val1")
f = self.query('GET/key.txt')
- self.assertTrue(f.read() == "val0")
+ self.assertTrue(f.read() == b"val0")
if __name__ == '__main__':
unittest.main()
self.s.connect((HOST, PORT))
def recv(self):
- out = ""
+ out = b""
while True:
try:
ret = self.s.recv(4096)
out += ret
def recv_until(self, limit):
- out = ""
+ out = b""
while not limit in out:
try:
out += self.s.recv(4096)
def __getitem__(self, chunk):
if chunk.start > self.len:
- return ""
+ return b""
if chunk.start + chunk.stop > self.len:
return self.char * (self.len - chunk.start)
return self.char * chunk.stop
def test_huge_url(self):
n = 1024*1024*1024 # 1GB query-string
- start = "GET /GET/x"
- end = " HTTP/1.0\r\n\r\n"
+ start = b"GET /GET/x"
+ end = b" HTTP/1.0\r\n\r\n"
ok = self.s.send(start)
- fail1 = self.s.send(LargeString("A", n))
+ fail1 = self.s.send(LargeString(b"A", n))
fail2 = self.s.send(end)
out = self.s.recv()
self.assertTrue(ok)
- self.assertTrue("400 Bad Request" in out)
+ self.assertTrue(b"400 Bad Request" in out)
def test_huge_upload(self):
n = 1024*1024*1024 # upload 1GB
- start = "PUT /SET/x HTTP/1.0\r\n"\
- + ("Content-Length: %d\r\n" % (n))\
- + "Expect: 100-continue\r\n\r\n"
+ start = b"PUT /SET/x HTTP/1.0\r\n"\
+ + (b"Content-Length: %d\r\n" % (n))\
+ + b"Expect: 100-continue\r\n\r\n"
ok = self.s.send(start)
- cont = self.s.recv_until("\r\n")
- fail = self.s.send(LargeString("A", n))
+ cont = self.s.recv_until(b"\r\n")
+ fail = self.s.send(LargeString(b"A", n))
self.assertTrue(ok)
- self.assertTrue("HTTP/1.1 100 Continue" in cont)
+ self.assertTrue(b"HTTP/1.1 100 Continue" in cont)
self.assertFalse(fail)
if __name__ == '__main__':
- unittest.main()
+ unittest.main(verbosity=5)