openstack/deb-python-eventlet
Code Issues Proposed changes

Python 3 compatibility fixes

Browse Source 
Closes GH-102
Closes GH-103
Closes GH-104
This commit is contained in:
Jakub Stasiak
2014-06-16 01:31:25 +01:00
committed by Sergey Shepelev
parent f37a87b1f8
commit 6afd8bdee2
17 changed files with 217 additions and 179 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
benchmarks/__init__.py
View File

@@ -1,7 +1,10 @@
import gc
import timeit
import random
from eventlet.support import six
def measure_best(repeat, iters,
common_setup='pass',
common_cleanup='pass',
@@ -16,9 +19,8 @@ def measure_best(repeat, iters,
t = timeit.Timer(func, setup=common_setup)
results[func].append(t.timeit(iters))
common_cleanup()
best_results = {}
for func, times in results.iteritems():
for func, times in six.iteritems(results):
best_results[func] = min(times)
return best_results

7
eventlet/green/BaseHTTPServer.py
View File

@@ -1,11 +1,14 @@
from eventlet import patcher
from eventlet.green import socket
from eventlet.green import SocketServer
from eventlet.support import six
patcher.inject('BaseHTTPServer',
patcher.inject(
'BaseHTTPServer' if six.PY2 else 'http.server',
globals(),
('socket', socket),
('SocketServer', SocketServer))
('SocketServer', SocketServer),
('socketserver', SocketServer))
del patcher

5
eventlet/green/SocketServer.py
View File

@@ -3,7 +3,10 @@ from eventlet import patcher
from eventlet.green import socket
from eventlet.green import select
from eventlet.green import threading
patcher.inject('SocketServer',
from eventlet.support import six
patcher.inject(
'SocketServer' if six.PY2 else 'socketserver',
globals(),
('socket', socket),
('select', select),

10
eventlet/green/profile.py
View File

@@ -41,6 +41,8 @@ import functools
from eventlet import greenthread
from eventlet import patcher
from eventlet.support import six
thread = patcher.original('thread') # non-monkeypatched module needed
@@ -154,7 +156,7 @@ class Profile(profile_orig.Profile):
return ContextWrapper
#Add automatic tasklet detection to the callbacks.
dispatch = dict([(key, ContextWrap(val)) for key,val in dispatch.iteritems()])
dispatch = dict([(key, ContextWrap(val)) for key, val in six.iteritems(dispatch)])
def TallyTimings(self):
oldtimings = self.sleeping
@@ -166,10 +168,10 @@ class Profile(profile_orig.Profile):
#we must keep the timings dicts separate for each tasklet, since it contains
#the 'ns' item, recursion count of each function in that tasklet. This is
#used in the Unwind dude.
for tasklet, (cur,timings) in oldtimings.iteritems():
for tasklet, (cur, timings) in six.iteritems(oldtimings):
self.Unwind(cur, timings)
for k,v in timings.iteritems():
for k, v in six.iteritems(timings):
if k not in self.timings:
self.timings[k] = v
else:
@@ -179,7 +181,7 @@ class Profile(profile_orig.Profile):
cc+=v[0]
tt+=v[2]
ct+=v[3]
for k1,v1 in v[4].iteritems():
for k1, v1 in six.iteritems(v[4]):
callers[k1] = callers.get(k1, 0)+v1
self.timings[k] = cc, ns, tt, ct, callers

5
eventlet/green/urllib.py
View File

@@ -13,6 +13,11 @@ except ImportError:
pass
patcher.inject('urllib', globals(), *to_patch)
try:
URLopener
except NameError:
patcher.inject('urllib.request', globals(), *to_patch)
# patch a bunch of things that have imports inside the
# function body; this is lame and hacky but I don't feel

2
eventlet/green/zmq.py
View File

@@ -40,7 +40,7 @@ class _QueueLock(object):
self._hub = hubs.get_hub()
def __nonzero__(self):
return self._count
return bool(self._count)
__bool__ = __nonzero__

8
eventlet/greenio.py
View File

@@ -553,16 +553,16 @@ try:
except ImportError:
# pyOpenSSL not installed, define exceptions anyway for convenience
class SSL(object):
class WantWriteError(object):
class WantWriteError(Exception):
pass
class WantReadError(object):
class WantReadError(Exception):
pass
class ZeroReturnError(object):
class ZeroReturnError(Exception):
pass
class SysCallError(object):
class SysCallError(Exception):
pass

4
eventlet/hubs/__init__.py
View File

@@ -1,6 +1,6 @@
import sys
import os
from eventlet.support import greenlets as greenlet
from eventlet.support import greenlets as greenlet, six
from eventlet import patcher
try:
@@ -77,7 +77,7 @@ def use_hub(mod=None):
mod = get_default_hub()
if hasattr(_threadlocal, 'hub'):
del _threadlocal.hub
if isinstance(mod, str):
if isinstance(mod, six.string_types):
assert mod.strip(), "Need to specify a hub"
if '.' in mod or ':' in mod:
modulename, _, classname = mod.strip().partition(':')

3
eventlet/hubs/pyevent.py
View File

@@ -5,6 +5,7 @@ import types
from eventlet.support import greenlets as greenlet, six
from eventlet.hubs.hub import BaseHub, FdListener, READ, WRITE
from eventlet.support import six
class event_wrapper(object):
@@ -127,7 +128,7 @@ class Hub(BaseHub):
listener.cb.delete()
def remove_descriptor(self, fileno):
for lcontainer in self.listeners.itervalues():
for lcontainer in six.itervalues(self.listeners):
listener = lcontainer.pop(fileno, None)
if listener:
try:

16
eventlet/websocket.py
View File

@@ -19,7 +19,7 @@ import eventlet
from eventlet import semaphore
from eventlet import wsgi
from eventlet.green import socket
from eventlet.support import get_errno
from eventlet.support import get_errno, six
# Python 2's utf8 decoding is more lenient than we'd like
# In order to pass autobahn's testsuite we need stricter validation
@@ -288,11 +288,11 @@ class WebSocket(object):
As per the dataframing section (5.3) for the websocket spec
"""
if isinstance(message, unicode):
if isinstance(message, six.text_type):
message = message.encode('utf-8')
elif not isinstance(message, str):
message = str(message)
packed = "\x00%s\xFF" % message
elif not isinstance(message, six.binary_type):
message = b'%s' % (message,)
packed = b"\x00%s\xFF" % message
return packed
def _parse_messages(self):
@@ -363,7 +363,7 @@ class WebSocket(object):
"""Sends the closing frame to the client, if required."""
if self.version == 76 and not self.websocket_closed:
try:
self.socket.sendall("\xff\x00")
self.socket.sendall(b"\xff\x00")
except SocketError:
# Sometimes, like when the remote side cuts off the connection,
# we don't care about this.
@@ -578,7 +578,7 @@ class RFC6455WebSocket(WebSocket):
def _pack_message(message, masked=False,
continuation=False, final=True, control_code=None):
is_text = False
if isinstance(message, unicode):
if isinstance(message, six.text_type):
message = message.encode('utf-8')
is_text = True
length = len(message)
@@ -634,7 +634,7 @@ class RFC6455WebSocket(WebSocket):
if self.version in (8, 13) and not self.websocket_closed:
if close_data is not None:
status, msg = close_data
if isinstance(msg, unicode):
if isinstance(msg, six.text_type):
msg = msg.encode('utf-8')
data = struct.pack('!H', status) + msg
else:

42
eventlet/wsgi.py
View File

@@ -56,6 +56,9 @@ class _AlreadyHandled(object):
def next(self):
raise StopIteration
__next__ = next
ALREADY_HANDLED = _AlreadyHandled()
@@ -144,13 +147,13 @@ class Input(object):
if use_readline and data[-1] == "\n":
break
else:
self.chunk_length = int(rfile.readline().split(";", 1)[0], 16)
self.chunk_length = int(rfile.readline().split(b";", 1)[0], 16)
self.position = 0
if self.chunk_length == 0:
rfile.readline()
except greenio.SSL.ZeroReturnError:
pass
return ''.join(response)
return b''.join(response)
def read(self, length=None):
if self.chunked_input:
@@ -268,7 +271,7 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
finally:
self.rfile = orig_rfile
content_length = self.headers.getheader('content-length')
content_length = self.headers.get('content-length')
if content_length:
try:
int(content_length)
@@ -356,7 +359,7 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
_writelines(towrite)
length[0] = length[0] + sum(map(len, towrite))
except UnicodeEncodeError:
self.server.log_message("Encountered non-ascii unicode while attempting to write wsgi response: %r" % [x for x in towrite if isinstance(x, unicode)])
self.server.log_message("Encountered non-ascii unicode while attempting to write wsgi response: %r" % [x for x in towrite if isinstance(x, six.text_type)])
self.server.log_message(traceback.format_exc())
_writelines(
["HTTP/1.1 500 Internal Server Error\r\n",
@@ -482,12 +485,15 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
if len(pq) > 1:
env['QUERY_STRING'] = pq[1]
if self.headers.typeheader is None:
env['CONTENT_TYPE'] = self.headers.type
else:
env['CONTENT_TYPE'] = self.headers.typeheader
ct = self.headers.get('content-type')
if ct is None:
try:
ct = self.headers.type
except AttributeError:
ct = self.headers.get_content_type()
env['CONTENT_TYPE'] = ct
length = self.headers.getheader('content-length')
length = self.headers.get('content-length')
if length:
env['CONTENT_LENGTH'] = length
env['SERVER_PROTOCOL'] = 'HTTP/1.0'
@@ -499,8 +505,14 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
env['REMOTE_PORT'] = str(self.client_address[1])
env['GATEWAY_INTERFACE'] = 'CGI/1.1'
for h in self.headers.headers:
k, v = h.split(':', 1)
try:
headers = self.headers.headers
except AttributeError:
headers = self.headers._headers
else:
headers = [h.split(':', 1) for h in headers]
for k, v in headers:
k = k.replace('-', '_').upper()
v = v.strip()
if k in env:
@@ -607,7 +619,7 @@ class Server(BaseHTTPServer.HTTPServer):
# set minimum_chunk_size before __init__ executes and we don't want to modify
# class variable
sock, address = sock_params
proto = types.InstanceType(self.protocol)
proto = new(self.protocol)
if self.minimum_chunk_size is not None:
proto.minimum_chunk_size = self.minimum_chunk_size
proto.capitalize_response_headers = self.capitalize_response_headers
@@ -624,6 +636,12 @@ class Server(BaseHTTPServer.HTTPServer):
self.log.write(message + '\n')
try:
new = types.InstanceType
except AttributeError:
new = lambda cls: cls.__new__(cls)
try:
import ssl
ACCEPT_EXCEPTIONS = (socket.error, ssl.SSLError)

4
tests/api_test.py
View File

@@ -42,7 +42,7 @@ class TestApi(TestCase):
def accept_once(listenfd):
try:
conn, addr = listenfd.accept()
fd = conn.makefile(mode='w')
fd = conn.makefile(mode='wb')
conn.close()
fd.write(b'hello\n')
fd.close()
@@ -53,7 +53,7 @@ class TestApi(TestCase):
api.spawn(accept_once, server)
client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
fd = client.makefile()
fd = client.makefile('rb')
client.close()
assert fd.readline() == b'hello\n'

18
tests/greenio_test.py
View File

@@ -62,11 +62,11 @@ class TestGreenSocket(LimitedTestCase):
# 2.x socket._fileobjects are odd: writes don't check
# whether the socket is closed or not, and you get an
# AttributeError during flush if it is closed
fd.write('a')
fd.write(b'a')
self.assertRaises(Exception, fd.flush)
else:
# 3.x io write to closed file-like pbject raises ValueError
self.assertRaises(ValueError, fd.write, 'a')
self.assertRaises(ValueError, fd.write, b'a')
def test_connect_timeout(self):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -453,7 +453,7 @@ class TestGreenSocket(LimitedTestCase):
wrap_wfile = s2.makefile('w')
eventlet.sleep(0.02)
wrap_wfile.write('hi')
wrap_wfile.write(b'hi')
s2.close()
evt.send(b'sent via event')
@@ -632,7 +632,7 @@ class TestGreenPipe(LimitedTestCase):
f.write(ch)
f.close()
one_line = "12345\n"
one_line = b"12345\n"
eventlet.spawn(sender, wf, one_line * 5)
for i in range(5):
line = rf.readline()
@@ -652,10 +652,10 @@ class TestGreenPipe(LimitedTestCase):
def writer():
eventlet.sleep(.1)
w.write('line\n')
w.write(b'line\n')
w.flush()
w.write('line\r\n')
w.write(b'line\r\n')
w.flush()
gt = eventlet.spawn(writer)
@@ -676,7 +676,7 @@ class TestGreenPipe(LimitedTestCase):
r = greenio.GreenPipe(r)
w = greenio.GreenPipe(w, 'w')
large_message = "".join([1024 * chr(i) for i in range(65)])
large_message = b"".join([1024 * chr(i) for i in range(65)])
def writer():
w.write(large_message)
@@ -698,7 +698,7 @@ class TestGreenPipe(LimitedTestCase):
self.assertEqual(f.tell(), 0)
f.seek(0, 2)
self.assertEqual(f.tell(), 0)
f.write('1234567890')
f.write(b'1234567890')
f.seek(0, 2)
self.assertEqual(f.tell(), 10)
f.seek(0)
@@ -719,7 +719,7 @@ class TestGreenPipe(LimitedTestCase):
def test_truncate(self):
f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024)
f.write('1234567890')
f.write(b'1234567890')
f.truncate(9)
self.assertEqual(f.tell(), 9)

4
tests/patcher_psycopg_test.py
View File

@@ -1,5 +1,7 @@
import os
from eventlet.support import six
from tests import patcher_test, skip_unless
from tests import get_database_auth
from tests.db_pool_test import postgres_requirement
@@ -45,7 +47,7 @@ class PatchingPsycopg(patcher_test.ProcessBase):
if isinstance(psycopg_auth, str):
dsn = psycopg_auth
else:
dsn = " ".join(["%s=%s" % (k, v) for k, v in psycopg_auth.iteritems()])
dsn = " ".join(["%s=%s" % (k, v) for k, v in six.iteritems(psycopg_auth)])
os.environ['PSYCOPG_TEST_DSN'] = dsn
self.write_to_tempfile("psycopg_patcher", psycopg_test_file)
output, lines = self.launch_subprocess('psycopg_patcher.py')

153
tests/wsgi_test.py
View File

@@ -84,10 +84,10 @@ def use_write(env, start_response):
if env['PATH_INFO'] == '/a':
write = start_response('200 OK', [('Content-type', 'text/plain'),
('Content-Length', '5')])
write('abcde')
write(b'abcde')
if env['PATH_INFO'] == '/b':
write = start_response('200 OK', [('Content-type', 'text/plain')])
write('abcde')
write(b'abcde')
return []
@@ -258,7 +258,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.flush()
result = fd.read()
fd.close()
@@ -271,10 +271,10 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
read_http(sock)
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
read_http(sock)
fd.close()
@@ -286,7 +286,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
cancel = eventlet.Timeout(1, RuntimeError)
self.assertRaises(TypeError, fd.read, "This shouldn't work")
@@ -298,13 +298,13 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
read_http(sock)
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
read_http(sock)
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
self.assertRaises(ConnectionClosed, read_http, sock)
fd.close()
@@ -332,8 +332,8 @@ class TestHttpd(_TestBase):
if result:
# windows closes the socket before the data is flushed,
# so we never get anything back
status = result.split(' ')[1]
self.assertEqual(status, '414')
status = result.split(b' ')[1]
self.assertEqual(status, b'414')
fd.close()
def test_007_get_arg(self):
@@ -342,7 +342,7 @@ class TestHttpd(_TestBase):
body = env['wsgi.input'].read()
a = cgi.parse_qs(body).get('a', [1])[0]
start_response('200 OK', [('Content-type', 'text/plain')])
return ['a is %s, body is %s' % (a, body)]
return [six.b('a is %s, body is %s' % (a, body))]
self.site.application = new_app
sock = eventlet.connect(
@@ -358,7 +358,7 @@ class TestHttpd(_TestBase):
fd.flush()
# send some junk after the actual request
fd.write('01234567890123456789')
fd.write(b'01234567890123456789')
result = read_http(sock)
self.assertEqual(result.body, 'a is a, body is a=a')
fd.close()
@@ -367,13 +367,13 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
result_200 = read_http(sock)
fd.write('GET /notexist HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET /notexist HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
read_http(sock)
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
result_test = read_http(sock)
self.assertEqual(result_200.status, result_test.status)
@@ -386,7 +386,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
assert 'Transfer-Encoding: chunked' in fd.read()
@@ -396,7 +396,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
assert 'Transfer-Encoding: chunked' not in fd.read()
@@ -406,7 +406,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
headers = ''
while True:
@@ -445,7 +445,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(('localhost', self.port))
sock = eventlet.wrap_ssl(sock)
sock.write('POST /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-length:3\r\n\r\nabc')
sock.write(b'POST /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-length:3\r\n\r\nabc')
result = sock.read(8192)
self.assertEqual(result[-3:], 'abc')
@@ -465,7 +465,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(('localhost', server_sock.getsockname()[1]))
sock = eventlet.wrap_ssl(sock)
sock.write('GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
sock.write(b'GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
result = sock.read(8192)
self.assertEqual(result[-4:], '\r\n\r\n')
@@ -475,7 +475,7 @@ class TestHttpd(_TestBase):
fd = sock.makefile('rw')
fd.write('PUT /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n'
'Transfer-Encoding: chunked\r\n\r\n'
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n')
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n'.encode())
fd.flush()
while True:
if fd.readline() == '\r\n':
@@ -487,7 +487,7 @@ class TestHttpd(_TestBase):
fd = sock.makefile('rw')
fd.write('PUT /b HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n'
'Transfer-Encoding: chunked\r\n\r\n'
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n')
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n'.encode())
fd.flush()
while True:
if fd.readline() == '\r\n':
@@ -499,7 +499,7 @@ class TestHttpd(_TestBase):
fd = sock.makefile('rw')
fd.write('PUT /c HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n'
'Transfer-Encoding: chunked\r\n\r\n'
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n')
'2\r\noh\r\n4\r\n hai\r\n0\r\n\r\n'.encode())
fd.flush()
while True:
if fd.readline() == '\r\n':
@@ -511,14 +511,14 @@ class TestHttpd(_TestBase):
self.site.application = use_write
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
result1 = read_http(sock)
assert 'content-length' in result1.headers_lower
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET /b HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET /b HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
result2 = read_http(sock)
assert 'transfer-encoding' in result2.headers_lower
@@ -535,7 +535,7 @@ class TestHttpd(_TestBase):
self.site.application = wsgi_app
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET /a HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
header_lines = []
while True:
@@ -575,7 +575,7 @@ class TestHttpd(_TestBase):
client = eventlet.connect(('localhost', sock.getsockname()[1]))
client = eventlet.wrap_ssl(client)
client.write('X') # non-empty payload so that SSL handshake occurs
client.write(b'X') # non-empty payload so that SSL handshake occurs
greenio.shutdown_safe(client)
client.close()
@@ -589,14 +589,14 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
fd.flush()
result1 = read_http(sock)
assert 'connection' in result1.headers_lower
self.assertEqual('keep-alive', result1.headers_lower['connection'])
# repeat request to verify connection is actually still open
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
fd.flush()
result2 = read_http(sock)
assert 'connection' in result2.headers_lower
@@ -619,7 +619,7 @@ class TestHttpd(_TestBase):
'Connection: close\r\n'
'Transfer-Encoding: chunked\r\n\r\n'
'2\r\noh\r\n'
'4\r\n hai\r\n0\r\n\r\n')
'4\r\n hai\r\n0\r\n\r\n'.encode())
fd.flush()
assert 'hello!' in fd.read()
@@ -655,7 +655,7 @@ class TestHttpd(_TestBase):
# do a single req/response to verify it's up
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.flush()
result = fd.read(1024)
fd.close()
@@ -674,7 +674,7 @@ class TestHttpd(_TestBase):
self.spawn_server(sock=server_sock)
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.flush()
result = fd.read(1024)
fd.close()
@@ -699,7 +699,7 @@ class TestHttpd(_TestBase):
fd.write('GET / HTTP/1.1\r\n'
'Host: localhost\r\n'
'Connection: close\r\n'
'\r\n\r\n')
'\r\n\r\n'.encode())
fd.flush()
assert '200 OK' in fd.read()
@@ -714,7 +714,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
fd.flush()
result = fd.read()
fd.close()
@@ -725,7 +725,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.0\r\nHost: localhost\r\nContent-length: argh\r\n\r\n')
fd.write(b'GET / HTTP/1.0\r\nHost: localhost\r\nContent-length: argh\r\n\r\n')
fd.flush()
result = fd.read()
fd.close()
@@ -745,12 +745,12 @@ class TestHttpd(_TestBase):
self.site.application = wsgi_app
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 1025\r\nExpect: 100-continue\r\n\r\n')
fd.write(b'PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 1025\r\nExpect: 100-continue\r\n\r\n')
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 417 Expectation Failed')
self.assertEqual(result.body, 'failure')
fd.write('PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 7\r\nExpect: 100-continue\r\n\r\ntesting')
fd.write(b'PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 7\r\nExpect: 100-continue\r\n\r\ntesting')
fd.flush()
header_lines = []
while True:
@@ -798,7 +798,7 @@ class TestHttpd(_TestBase):
def test_026_log_format(self):
self.spawn_server(log_format="HI %(request_line)s HI")
sock = eventlet.connect(('localhost', self.port))
sock.sendall('GET /yo! HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.sendall(b'GET /yo! HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.recv(1024)
sock.close()
assert '\nHI GET /yo! HTTP/1.1 HI\n' in self.logfile.getvalue(), self.logfile.getvalue()
@@ -810,7 +810,7 @@ class TestHttpd(_TestBase):
self.site.application = chunked_app
sock = eventlet.connect(('localhost', self.port))
sock.sendall('GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
result = read_http(sock)
self.assertEqual(result.headers_lower['connection'], 'close')
@@ -822,7 +822,7 @@ class TestHttpd(_TestBase):
self.spawn_server(minimum_chunk_size=start_size * 2)
sock = eventlet.connect(('localhost', self.port))
sock.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
read_http(sock)
self.assertEqual(wsgi.HttpProtocol.minimum_chunk_size, start_size)
@@ -835,7 +835,7 @@ class TestHttpd(_TestBase):
self.site.application = chunked_fail_app
sock = eventlet.connect(('localhost', self.port))
sock.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 200 OK')
@@ -855,7 +855,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(
('localhost', self.port))
sock.sendall('GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
sock.sendall(b'GET / HTTP/1.0\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n')
result = read_http(sock)
self.assertEqual(result.headers_lower['connection'], 'close')
@@ -892,6 +892,7 @@ class TestHttpd(_TestBase):
pass
except Exception as e:
errored[0] = 'SSL handshake error raised exception %s.' % e
raise
for data in ('', 'GET /non-ssl-request HTTP/1.0\r\n\r\n'):
srv_sock = eventlet.wrap_ssl(
eventlet.listen(('localhost', 0)),
@@ -901,7 +902,7 @@ class TestHttpd(_TestBase):
g = eventlet.spawn_n(server, srv_sock)
client = eventlet.connect(('localhost', port))
if data: # send non-ssl request
client.sendall(data)
client.sendall(data.encode())
else: # close sock prematurely
client.close()
eventlet.sleep(0) # let context switch back to server
@@ -909,7 +910,7 @@ class TestHttpd(_TestBase):
# make another request to ensure the server's still alive
try:
client = ssl.wrap_socket(eventlet.connect(('localhost', port)))
client.write('GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
client.write(b'GET / HTTP/1.0\r\nHost: localhost\r\n\r\n')
result = client.read()
assert result.startswith('HTTP'), result
assert result.endswith('hello world')
@@ -942,7 +943,7 @@ class TestHttpd(_TestBase):
self.site.application = one_posthook_app
sock = eventlet.connect(('localhost', self.port))
fp = sock.makefile('rw')
fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.flush()
self.assertEqual(fp.readline(), 'HTTP/1.1 200 OK\r\n')
fp.close()
@@ -965,7 +966,7 @@ class TestHttpd(_TestBase):
self.site.application = two_posthook_app
sock = eventlet.connect(('localhost', self.port))
fp = sock.makefile('rw')
fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.flush()
self.assertEqual(fp.readline(), 'HTTP/1.1 200 OK\r\n')
fp.close()
@@ -978,7 +979,7 @@ class TestHttpd(_TestBase):
request = 'GET / HTTP/1.0\r\nHost: localhost\r\nLong: %s\r\n\r\n' % \
('a' * 10000)
fd = sock.makefile('rw')
fd.write(request)
fd.write(request.encode())
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.0 400 Header Line Too Long')
@@ -988,8 +989,8 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(('localhost', self.port))
headers = 'Name: Value\r\n' * 5050
request = 'GET / HTTP/1.0\r\nHost: localhost\r\n%s\r\n\r\n' % headers
fd = sock.makefile('rw')
fd.write(request)
fd = sock.makefile('rwb')
fd.write(request.encode())
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.0 400 Headers Too Large')
@@ -1015,8 +1016,8 @@ class TestHttpd(_TestBase):
'Content-Length: %i\r\n\r\n%s'
) % (len(upload_data), upload_data)
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write(request)
fd = sock.makefile('rwb')
fd.write(request.encode())
fd.flush()
result = read_http(sock)
self.assertEqual(result.body, upload_data)
@@ -1033,7 +1034,7 @@ class TestHttpd(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
response = fd.read().split('\r\n')
headers = []
@@ -1086,7 +1087,7 @@ class TestHttpd(_TestBase):
expected_body])
# start PUT-ing some chunked data but close prematurely
sock = eventlet.connect(('127.0.0.1', self.port))
sock.sendall(data)
sock.sendall(data.encode())
sock.close()
# the test passes if we successfully get here, and read all the data
# in spite of the early close
@@ -1099,7 +1100,7 @@ class TestHttpd(_TestBase):
self.site.application = wsgi_app
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 500 Internal Server Error')
@@ -1114,7 +1115,7 @@ class TestHttpd(_TestBase):
self.site.application = wsgi_app
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 500 Internal Server Error')
@@ -1129,7 +1130,7 @@ class TestHttpd(_TestBase):
self.site.application = wsgi_app
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET /a*b@%40%233 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET /a*b@%40%233 HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 200 OK')
@@ -1148,7 +1149,7 @@ class TestHttpd(_TestBase):
try:
wsgi.server(sock=sock, log=log, site=Site())
except ValueError:
log.write('broked')
log.write(b'broken')
self.spawn_thread(run_server)
@@ -1168,7 +1169,7 @@ class TestHttpd(_TestBase):
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
result1 = read_http(sock)
self.assertEqual(result1.status, 'HTTP/1.1 500 Internal Server Error')
@@ -1181,7 +1182,7 @@ class TestHttpd(_TestBase):
self.site.application = crasher
sock = eventlet.connect(('localhost', self.port))
fd = sock.makefile('w')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
result2 = read_http(sock)
self.assertEqual(result2.status, 'HTTP/1.1 500 Internal Server Error')
@@ -1205,7 +1206,7 @@ class TestHttpd(_TestBase):
def make_request():
sock = eventlet.connect(server_sock.getsockname())
sock.send('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.send(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.close()
request_thread = eventlet.spawn(make_request)
@@ -1223,7 +1224,7 @@ class TestHttpd(_TestBase):
os.path.dirname(os.path.abspath(__file__)),
'wsgi_test_conntimeout.py')
output = run_python(testcode_path)
sections = output.split("SEPERATOR_SENTINEL")
sections = output.split(b"SEPERATOR_SENTINEL")
# first section is empty
self.assertEqual(3, len(sections), output)
# if the "BOOM" check fails, it's because our timeout didn't happen
@@ -1239,7 +1240,7 @@ class TestHttpd(_TestBase):
def test_server_socket_timeout(self):
self.spawn_server(socket_timeout=0.1)
sock = eventlet.connect(('localhost', self.port))
sock.send('GET / HTTP/1.1\r\n')
sock.send(b'GET / HTTP/1.1\r\n')
eventlet.sleep(0.1)
try:
read_http(sock)
@@ -1260,7 +1261,7 @@ class TestHttpd(_TestBase):
self.spawn_server(site=wsgi_app, capitalize_response_headers=False)
sock = eventlet.connect(('localhost', self.port))
sock.sendall('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
sock.sendall(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
result = read_http(sock)
sock.close()
self.assertEqual(result.status, 'HTTP/1.1 200 oK')
@@ -1310,13 +1311,13 @@ class IterableAlreadyHandledTest(_TestBase):
('localhost', self.port))
fd = sock.makefile('rw')
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd.flush()
response_line, headers = read_headers(sock)
self.assertEqual(response_line, 'HTTP/1.1 200 OK\r\n')
assert 'connection' not in headers
fd.write('GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.write(b'GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
fd.flush()
result = read_http(sock)
self.assertEqual(result.status, 'HTTP/1.1 200 OK')
@@ -1354,7 +1355,7 @@ class TestChunkedInput(_TestBase):
response.append(x)
elif pi == "/ping":
input.read()
response.append("pong")
response.append(b"pong")
elif pi.startswith("/yield_spaces"):
if pi.endswith('override_min'):
env['eventlet.minimum_write_chunk_size'] = 1
@@ -1399,7 +1400,7 @@ class TestChunkedInput(_TestBase):
return self.chunk_encode(["this", " is ", "chunked", "\nline", " 2", "\n", "line3", ""], dirt=dirt)
def ping(self, fd):
fd.sendall("GET /ping HTTP/1.1\r\n\r\n")
fd.sendall(b"GET /ping HTTP/1.1\r\n\r\n")
self.assertEqual(read_http(fd).body, "pong")
def test_short_read_with_content_length(self):
@@ -1407,7 +1408,7 @@ class TestChunkedInput(_TestBase):
req = "POST /short-read HTTP/1.1\r\ntransfer-encoding: Chunked\r\nContent-Length:1000\r\n\r\n" + body
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd)
@@ -1417,7 +1418,7 @@ class TestChunkedInput(_TestBase):
body = self.body()
req = "POST /short-read HTTP/1.1\r\ntransfer-encoding: Chunked\r\nContent-Length:0\r\n\r\n" + body
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd)
@@ -1428,7 +1429,7 @@ class TestChunkedInput(_TestBase):
req = "POST /short-read HTTP/1.1\r\ntransfer-encoding: Chunked\r\n\r\n" + body
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd)
@@ -1439,7 +1440,7 @@ class TestChunkedInput(_TestBase):
req = "POST /ping HTTP/1.1\r\ntransfer-encoding: Chunked\r\n\r\n" + body
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
self.assertEqual(read_http(fd).body, "pong")
self.ping(fd)
@@ -1450,14 +1451,14 @@ class TestChunkedInput(_TestBase):
req = "POST /lines HTTP/1.1\r\nContent-Length: %s\r\ntransfer-encoding: Chunked\r\n\r\n%s" % (len(body), body)
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
self.assertEqual(read_http(fd).body, 'this is chunked\nline 2\nline3')
fd.close()
def test_chunked_readline_wsgi_override_minimum_chunk_size(self):
fd = self.connect()
fd.sendall("POST /yield_spaces/override_min HTTP/1.1\r\nContent-Length: 0\r\n\r\n")
fd.sendall(b"POST /yield_spaces/override_min HTTP/1.1\r\nContent-Length: 0\r\n\r\n")
resp_so_far = ''
with eventlet.Timeout(.1):
@@ -1482,7 +1483,7 @@ class TestChunkedInput(_TestBase):
def test_chunked_readline_wsgi_not_override_minimum_chunk_size(self):
fd = self.connect()
fd.sendall("POST /yield_spaces HTTP/1.1\r\nContent-Length: 0\r\n\r\n")
fd.sendall(b"POST /yield_spaces HTTP/1.1\r\nContent-Length: 0\r\n\r\n")
resp_so_far = ''
try:
@@ -1513,7 +1514,7 @@ class TestChunkedInput(_TestBase):
req = "POST /short-read HTTP/1.1\r\ntransfer-encoding: Chunked\r\n\r\n" + body
fd = self.connect()
fd.sendall(req)
fd.sendall(req.encode())
fd.close()
eventlet.sleep(0.0)
finally:

6
tests/wsgi_test_conntimeout.py
View File

@@ -93,7 +93,7 @@ class ExplodingConnectionWrap(object):
return file_obj
class ExplodingSocketFile(socket._fileobject):
class ExplodingSocketFile(eventlet.greenio._fileobject):
def __init__(self, sock, mode='rb', bufsize=-1, close=False):
super(self.__class__, self).__init__(sock, mode, bufsize, close)
@@ -137,7 +137,7 @@ if __name__ == '__main__':
sock1 = eventlet.connect(server_addr)
sock1.settimeout(0.1)
fd1 = sock1.makefile('rw')
fd1.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd1.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd1.flush()
tests.wsgi_test.read_http(sock1)
@@ -147,7 +147,7 @@ if __name__ == '__main__':
sock_wrap.arm()
# req #2 - old conn, post-arm - timeout
fd1.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd1.write(b'GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fd1.flush()
try:
tests.wsgi_test.read_http(sock1)

99
tests/zmq_test.py
View File

@@ -8,9 +8,8 @@ try:
from eventlet.green import zmq
except ImportError:
zmq = {} # for systems lacking zmq, skips tests instead of barfing
RECV_ON_CLOSED_SOCKET_ERRNOS = (zmq.ENOTSUP, zmq.ENOTSOCK)
else:
RECV_ON_CLOSED_SOCKET_ERRNOS = (zmq.ENOTSUP, zmq.ENOTSOCK)
def zmq_supported(_):
@@ -90,9 +89,9 @@ class TestUpstreamDownStream(LimitedTestCase):
done.send('done')
spawn(rx)
req.send('test')
req.send(b'test')
done.wait()
self.assertEqual(msg['res'], 'test')
self.assertEqual(msg['res'], b'test')
@skip_unless(zmq_supported)
def test_close_socket_raises_enotsup(self):
@@ -101,7 +100,7 @@ class TestUpstreamDownStream(LimitedTestCase):
rep.close()
req.close()
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, rep.recv)
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, 'test')
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, b'test')
@skip_unless(zmq_supported)
def test_close_xsocket_raises_enotsup(self):
@@ -110,7 +109,7 @@ class TestUpstreamDownStream(LimitedTestCase):
rep.close()
req.close()
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, rep.recv)
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, 'test')
self.assertRaisesErrno(RECV_ON_CLOSED_SOCKET_ERRNOS, req.send, b'test')
@skip_unless(zmq_supported)
def test_send_1k_req_rep(self):
@@ -120,19 +119,19 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx():
tx_i = 0
req.send(str(tx_i))
while req.recv() != 'done':
req.send(str(tx_i).encode())
while req.recv() != b'done':
tx_i += 1
req.send(str(tx_i))
req.send(str(tx_i).encode())
done.send(0)
def rx():
while True:
rx_i = rep.recv()
if rx_i == "1000":
rep.send('done')
if rx_i == b"1000":
rep.send(b'done')
break
rep.send('i')
rep.send(b'i')
spawn(tx)
spawn(rx)
final_i = done.wait()
@@ -149,12 +148,12 @@ class TestUpstreamDownStream(LimitedTestCase):
tx_i = 0
while tx_i <= 1000:
tx_i += 1
down.send(str(tx_i))
down.send(str(tx_i).encode())
def rx():
while True:
rx_i = up.recv()
if rx_i == "1000":
if rx_i == b"1000":
done.send(0)
break
spawn(tx)
@@ -171,9 +170,9 @@ class TestUpstreamDownStream(LimitedTestCase):
addr = 'tcp://127.0.0.1:%s' % port
sub1.connect(addr)
sub2.connect(addr)
sub_all.setsockopt(zmq.SUBSCRIBE, '')
sub1.setsockopt(zmq.SUBSCRIBE, 'sub1')
sub2.setsockopt(zmq.SUBSCRIBE, 'sub2')
sub_all.setsockopt(zmq.SUBSCRIBE, b'')
sub1.setsockopt(zmq.SUBSCRIBE, b'sub1')
sub2.setsockopt(zmq.SUBSCRIBE, b'sub2')
sub_all_done = event.Event()
sub1_done = event.Event()
@@ -186,7 +185,7 @@ class TestUpstreamDownStream(LimitedTestCase):
while count < msg_count:
msg = sock.recv()
sleep()
if 'LAST' in msg:
if b'LAST' in msg:
break
count += 1
@@ -194,11 +193,11 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx(sock):
for i in range(1, 1001):
msg = "sub%s %s" % ([2, 1][i % 2], i)
msg = ("sub%s %s" % ([2, 1][i % 2], i)).encode()
sock.send(msg)
sleep()
sock.send('sub1 LAST')
sock.send('sub2 LAST')
sock.send(b'sub1 LAST')
sock.send(b'sub2 LAST')
spawn(rx, sub_all, sub_all_done)
spawn(rx, sub1, sub1_done)
@@ -214,35 +213,35 @@ class TestUpstreamDownStream(LimitedTestCase):
@skip_unless(zmq_supported)
def test_change_subscription(self):
pub, sub, port = self.create_bound_pair(zmq.PUB, zmq.SUB)
sub.setsockopt(zmq.SUBSCRIBE, 'test')
sub.setsockopt(zmq.SUBSCRIBE, b'test')
sleep(0.2)
sub_done = event.Event()
def rx(sock, done_evt):
count = 0
sub = 'test'
sub = b'test'
while True:
msg = sock.recv()
sleep()
if 'DONE' in msg:
if b'DONE' in msg:
break
if 'LAST' in msg and sub == 'test':
sock.setsockopt(zmq.UNSUBSCRIBE, 'test')
sock.setsockopt(zmq.SUBSCRIBE, 'done')
sub = 'done'
if b'LAST' in msg and sub == b'test':
sock.setsockopt(zmq.UNSUBSCRIBE, b'test')
sock.setsockopt(zmq.SUBSCRIBE, b'done')
sub = b'done'
count += 1
done_evt.send(count)
def tx(sock):
for i in range(1, 101):
msg = "test %s" % i
msg = ("test %s" % i).encode()
if i != 50:
sock.send(msg)
else:
sock.send('test LAST')
sock.send(b'test LAST')
sleep()
sock.send('done DONE')
sock.send(b'done DONE')
spawn(rx, sub, sub_done)
spawn(tx, pub)
@@ -253,20 +252,20 @@ class TestUpstreamDownStream(LimitedTestCase):
@skip_unless(zmq_supported)
def test_recv_multipart_bug68(self):
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
msg = ['']
msg = [b'']
req.send_multipart(msg)
recieved_msg = rep.recv_multipart()
self.assertEqual(recieved_msg, msg)
# Send a message back the other way
msg2 = [""]
msg2 = [b""]
rep.send_multipart(msg2, copy=False)
# When receiving a copy it's a zmq.core.message.Message you get back
recieved_msg = req.recv_multipart(copy=False)
# So it needs to be converted to a string
# I'm calling str(m) consciously here; Message has a .data attribute
# but it's private __str__ appears to be the way to go
self.assertEqual([str(m) for m in recieved_msg], msg2)
self.assertEqual([m.bytes for m in recieved_msg], msg2)
@skip_unless(zmq_supported)
def test_recv_noblock_bug76(self):
@@ -289,20 +288,20 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx():
tx_i = 0
while tx_i <= 1000:
sender.send(str(tx_i))
sender.send(str(tx_i).encode())
tx_i += 1
def rx():
while True:
rx_i = receiver.recv()
if rx_i == "1000":
if rx_i == b"1000":
for i in range(num_recvs):
receiver.send('done%d' % i)
receiver.send(('done%d' % i).encode())
sleep()
return
for i in range(num_recvs):
spawn(slow_rx, done_evts[i], "done%d" % i)
spawn(slow_rx, done_evts[i], ("done%d" % i).encode())
spawn(tx)
spawn(rx)
@@ -324,20 +323,22 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx():
tx_i = 0
while tx_i <= 1000:
sender.send_multipart([str(tx_i), '1', '2', '3'])
sender.send_multipart([str(tx_i).encode(), b'1', b'2', b'3'])
tx_i += 1
def rx():
while True:
rx_i = receiver.recv_multipart()
if rx_i == ["1000", '1', '2', '3']:
if rx_i == [b"1000", b'1', b'2', b'3']:
for i in range(num_recvs):
receiver.send_multipart(['done%d' % i, 'a', 'b', 'c'])
receiver.send_multipart([
('done%d' % i).encode(), b'a', b'b', b'c'])
sleep()
return
for i in range(num_recvs):
spawn(slow_rx, done_evts[i], ["done%d" % i, 'a', 'b', 'c'])
spawn(slow_rx, done_evts[i], [
("done%d" % i).encode(), b'a', b'b', b'c'])
spawn(tx)
spawn(rx)
@@ -367,7 +368,7 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx():
tx_i = 0
while tx_i <= 1000:
sender.send(str(tx_i))
sender.send(str(tx_i).encode())
tx_i += 1
done.send(0)
@@ -405,7 +406,7 @@ class TestUpstreamDownStream(LimitedTestCase):
self.assertEqual(len(sock_map), 1)
events = sock1.getsockopt(zmq.EVENTS)
self.assertEqual(events & zmq.POLLOUT, zmq.POLLOUT)
sock1.send('')
sock1.send(b'')
poll_in = zmq.Poller()
poll_in.register(sock2, zmq.POLLIN)
@@ -440,16 +441,16 @@ class TestUpstreamDownStream(LimitedTestCase):
Same https://bitbucket.org/eventlet/eventlet/issue/128
"""
pub, sub, _port = self.create_bound_pair(zmq.PUB, zmq.SUB)
sub.setsockopt(zmq.SUBSCRIBE, "")
sub.setsockopt(zmq.SUBSCRIBE, b"")
sleep()
pub.send('test_send')
pub.send(b'test_send')
check_idle_cpu_usage(0.2, 0.1)
sender, receiver, _port = self.create_bound_pair(zmq.DEALER, zmq.DEALER)
sleep()
sender.send('test_recv')
sender.send(b'test_recv')
msg = receiver.recv()
self.assertEqual(msg, 'test_recv')
self.assertEqual(msg, b'test_recv')
check_idle_cpu_usage(0.2, 0.1)