Permalink
Browse files

Fixes for importing tests.

  • Loading branch information...
1 parent bc27dda commit f22bbc2f21411ab6a75db7599e991c99dceaba2c @takluyver takluyver committed Jan 22, 2012
Showing with 27 additions and 20 deletions.
  1. +10 −5 dummyserver/app.py
  2. +2 −1 dummyserver/server.py
  3. +7 −6 test/benchmark.py
  4. +5 −4 test/test_response.py
  5. +3 −4 test/with_dummyserver/test_https.py
View
@@ -1,12 +1,17 @@
+from __future__ import print_function
+
import gzip
import logging
import sys
import time
import zlib
-from urlparse import urlsplit
+try:
+ from urllib.parse import urlsplit
+except ImportError:
+ from urlparse import urlsplit
from cgi import FieldStorage
-from StringIO import StringIO
+from io import BytesIO
from webob import Request, Response, exc
@@ -47,9 +52,9 @@ def set_up(self, request):
test_type = request.params.get('test_type')
test_id = request.params.get('test_id')
if test_id:
- print '\nNew test %s: %s' % (test_type, test_id)
+ print('\nNew test %s: %s' % (test_type, test_id))
else:
- print '\nNew test %s' % test_type
+ print('\nNew test %s' % test_type)
return Response("Dummy server is ready!")
def specific_method(self, request):
@@ -117,7 +122,7 @@ def encodingrequest(self, request):
headers = {}
if 'gzip' in encoding:
headers = {'Content-Encoding': 'gzip'}
- file_ = StringIO()
+ file_ = BytesIO()
gzip.GzipFile('', mode='w', fileobj=file_).write(data)
data = file_.getvalue()
elif 'deflate' in encoding:
@@ -3,6 +3,7 @@
"""
Dummy server used for unit testing.
"""
+from __future__ import print_function
import logging
import os
@@ -136,7 +137,7 @@ def make_server_thread(**kw):
if len(sys.argv) > 1:
url = sys.argv[1]
- print "Starting server at: %s" % url
+ print("Starting server at: %s" % url)
scheme, host, port = get_host(url)
make_server(scheme=scheme, host=host, port=port)
View
@@ -4,6 +4,7 @@
Really simple rudimentary benchmark to compare ConnectionPool versus standard
urllib to demonstrate the usefulness of connection re-using.
"""
+from __future__ import print_function
import sys
import time
@@ -40,7 +41,7 @@ def urllib_get(url_list):
now = time.time()
r = urllib.urlopen(url)
elapsed = time.time() - now
- print "Got in %0.3f: %s" % (elapsed, url)
+ print("Got in %0.3f: %s" % (elapsed, url))
def pool_get(url_list):
@@ -50,22 +51,22 @@ def pool_get(url_list):
now = time.time()
r = pool.get_url(url)
elapsed = time.time() - now
- print "Got in %0.3fs: %s" % (elapsed, url)
+ print("Got in %0.3fs: %s" % (elapsed, url))
if __name__ == '__main__':
- print "Running pool_get ..."
+ print("Running pool_get ...")
now = time.time()
pool_get(TO_DOWNLOAD)
pool_elapsed = time.time() - now
- print "Running urllib_get ..."
+ print("Running urllib_get ...")
now = time.time()
urllib_get(TO_DOWNLOAD)
urllib_elapsed = time.time() - now
- print "Completed pool_get in %0.3fs" % pool_elapsed
- print "Completed urllib_get in %0.3fs" % urllib_elapsed
+ print("Completed pool_get in %0.3fs" % pool_elapsed)
+ print("Completed urllib_get in %0.3fs" % urllib_elapsed)
"""
@@ -1,6 +1,6 @@
import unittest
-from StringIO import StringIO
+from io import BytesIO, StringIO
from urllib3.response import HTTPResponse
@@ -24,12 +24,13 @@ def test_no_preload(self):
self.assertEqual(fp.tell(), fp.len)
def test_decode_deflate(self):
- data = 'foo'.encode('zlib')
+ import zlib
+ data = zlib.compress(b'foo')
- fp = StringIO(data)
+ fp = BytesIO(data)
r = HTTPResponse(fp, headers={'content-encoding': 'deflate'})
- self.assertEqual(r.data, 'foo')
+ self.assertEqual(r.data, b'foo')
if __name__ == '__main__':
@@ -14,7 +14,6 @@
log.setLevel(logging.NOTSET)
log.addHandler(logging.StreamHandler(sys.stdout))
-
class TestHTTPS(HTTPSDummyServerTestCase):
def setUp(self):
self._pool = HTTPSConnectionPool(self.host, self.port)
@@ -34,15 +33,15 @@ def test_verified(self):
try:
https_pool.request('GET', '/')
self.fail("Didn't raise SSL error with no CA")
- except SSLError, e:
+ except SSLError as e:
self.assertTrue('No root certificates' in str(e))
https_pool.ca_certs = DEFAULT_CA_BAD
try:
https_pool.request('GET', '/')
self.fail("Didn't raise SSL error with wrong CA")
- except SSLError, e:
+ except SSLError as e:
self.assertTrue('certificate verify failed' in str(e))
https_pool.ca_certs = DEFAULT_CA
@@ -55,7 +54,7 @@ def test_verified(self):
try:
https_fail_pool.request('GET', '/')
self.fail("Didn't raise SSL invalid common name")
- except SSLError, e:
+ except SSLError as e:
self.assertTrue("doesn't match" in str(e))

0 comments on commit f22bbc2

Please sign in to comment.