openstack/deb-python-eventlet
Code Issues Proposed changes

python3 compatibility

Browse Source 
- __next__ for iterator interface
- six.next() to get next item
- list(dict.keys())
- popen2.popen4 -> subprocess
- s2b -> b"..." literals
- deprecated assertEquals -> assertEqual
- hub_test test_fork using run_python
- 1L -> 1 long literal
- many PEP-8 fixes
This commit is contained in:
Sergey Shepelev
2014-04-24 17:51:27 +04:00
parent c467ded4a6
commit cbd404d56e
28 changed files with 564 additions and 477 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
eventlet/greenio.py
View File

@@ -9,10 +9,10 @@ import warnings
from eventlet.support import get_errno, six from eventlet.support import get_errno, six
from eventlet.hubs import trampoline from eventlet.hubs import trampoline
BUFFER_SIZE = 4096
__all__ = ['GreenSocket', 'GreenPipe', 'shutdown_safe'] __all__ = ['GreenSocket', 'GreenPipe', 'shutdown_safe']
BUFFER_SIZE = 4096
CONNECT_ERR = set((errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK)) CONNECT_ERR = set((errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK))
CONNECT_SUCCESS = set((0, errno.EISCONN)) CONNECT_SUCCESS = set((0, errno.EISCONN))
if sys.platform[:3] == "win": if sys.platform[:3] == "win":

13
eventlet/greenpool.py
View File

@@ -11,6 +11,7 @@ __all__ = ['GreenPool', 'GreenPile']
DEBUG = True DEBUG = True
class GreenPool(object): class GreenPool(object):
"""The GreenPool class is a pool of green threads. """The GreenPool class is a pool of green threads.
""" """
@@ -102,7 +103,8 @@ class GreenPool(object):
self._spawn_n_impl(function, args, kwargs, None) self._spawn_n_impl(function, args, kwargs, None)
else: else:
self.sem.acquire() self.sem.acquire()
g = greenthread.spawn_n(self._spawn_n_impl, g = greenthread.spawn_n(
self._spawn_n_impl,
function, args, kwargs, True) function, args, kwargs, True)
if not self.coroutines_running: if not self.coroutines_running:
self.no_coros_running = event.Event() self.no_coros_running = event.Event()
@@ -111,8 +113,8 @@ class GreenPool(object):
def waitall(self): def waitall(self):
"""Waits until all greenthreads in the pool are finished working.""" """Waits until all greenthreads in the pool are finished working."""
assert greenthread.getcurrent() not in self.coroutines_running, \ assert greenthread.getcurrent() not in self.coroutines_running, \
"Calling waitall() from within one of the "\ "Calling waitall() from within one of the " \
"GreenPool's greenthreads will never terminate." "GreenPool's greenthreads will never terminate."
if self.running(): if self.running():
self.no_coros_running.wait() self.no_coros_running.wait()
@@ -197,7 +199,7 @@ class GreenPile(object):
def spawn(self, func, *args, **kw): def spawn(self, func, *args, **kw):
"""Runs *func* in its own green thread, with the result available by """Runs *func* in its own green thread, with the result available by
iterating over the GreenPile object.""" iterating over the GreenPile object."""
self.used = True self.used = True
self.counter += 1 self.counter += 1
try: try:
gt = self.pool.spawn(func, *args, **kw) gt = self.pool.spawn(func, *args, **kw)
@@ -218,6 +220,8 @@ class GreenPile(object):
return self.waiters.get().wait() return self.waiters.get().wait()
finally: finally:
self.counter -= 1 self.counter -= 1
__next__ = next
# this is identical to GreenPile but it blocks on spawn if the results # this is identical to GreenPile but it blocks on spawn if the results
# aren't consumed, and it doesn't generate its own StopIteration exception, # aren't consumed, and it doesn't generate its own StopIteration exception,
@@ -236,3 +240,4 @@ class GreenMap(GreenPile):
return val return val
finally: finally:
self.counter -= 1 self.counter -= 1
__next__ = next

11
eventlet/hubs/selects.py
View File

@@ -1,7 +1,7 @@
import sys import sys
import errno import errno
from eventlet import patcher from eventlet import patcher
from eventlet.support import get_errno, clear_sys_exc_info from eventlet.support import get_errno, clear_sys_exc_info, six
select = patcher.original('select') select = patcher.original('select')
time = patcher.original('time') time = patcher.original('time')
@@ -12,12 +12,14 @@ try:
except AttributeError: except AttributeError:
BAD_SOCK = set((errno.EBADF,)) BAD_SOCK = set((errno.EBADF,))
class Hub(BaseHub): class Hub(BaseHub):
def _remove_bad_fds(self): def _remove_bad_fds(self):
""" Iterate through fds, removing the ones that are bad per the """ Iterate through fds, removing the ones that are bad per the
operating system. operating system.
""" """
for fd in self.listeners[READ].keys() + self.listeners[WRITE].keys(): all_fds = list(self.listeners[READ]) + list(self.listeners[WRITE])
for fd in all_fds:
try: try:
select.select([fd], [], [], 0) select.select([fd], [], [], 0)
except select.error as e: except select.error as e:
@@ -31,8 +33,9 @@ class Hub(BaseHub):
if seconds: if seconds:
time.sleep(seconds) time.sleep(seconds)
return return
all_fds = list(readers) + list(writers)
try: try:
r, w, er = select.select(readers.keys(), writers.keys(), readers.keys() + writers.keys(), seconds) r, w, er = select.select(readers.keys(), writers.keys(), all_fds, seconds)
except select.error as e: except select.error as e:
if get_errno(e) == errno.EINTR: if get_errno(e) == errno.EINTR:
return return
@@ -45,7 +48,7 @@ class Hub(BaseHub):
for fileno in er: for fileno in er:
readers.get(fileno, noop).cb(fileno) readers.get(fileno, noop).cb(fileno)
writers.get(fileno, noop).cb(fileno) writers.get(fileno, noop).cb(fileno)
for listeners, events in ((readers, r), (writers, w)): for listeners, events in ((readers, r), (writers, w)):
for fileno in events: for fileno in events:
try: try:

24
eventlet/pool.py
View File

@@ -2,6 +2,7 @@ from __future__ import print_function
from eventlet import coros, proc, api from eventlet import coros, proc, api
from eventlet.semaphore import Semaphore from eventlet.semaphore import Semaphore
from eventlet.support import six
import warnings import warnings
warnings.warn( warnings.warn(
@@ -190,6 +191,7 @@ class Pool(object):
millions of items.) millions of items.)
>>> from eventlet import coros >>> from eventlet import coros
>>> from eventlet.support import six
>>> import string >>> import string
>>> pool = coros.CoroutinePool(max_size=5) >>> pool = coros.CoroutinePool(max_size=5)
>>> pausers = [coros.Event() for x in range(2)] >>> pausers = [coros.Event() for x in range(2)]
@@ -206,39 +208,39 @@ class Pool(object):
items individually to illustrate timing) items individually to illustrate timing)
>>> step = iter(pool.generate_results(quicktask, string.ascii_lowercase)) >>> step = iter(pool.generate_results(quicktask, string.ascii_lowercase))
>>> print(step.next()) >>> print(six.next(step))
returning a returning a
returning b returning b
returning c returning c
a a
>>> print(step.next()) >>> print(six.next(step))
b b
>>> print(step.next()) >>> print(six.next(step))
c c
>>> print(step.next()) >>> print(six.next(step))
returning d returning d
returning e returning e
returning f returning f
d d
>>> pausers[0].send("A") >>> pausers[0].send("A")
>>> print(step.next()) >>> print(six.next(step))
e e
>>> print(step.next()) >>> print(six.next(step))
f f
>>> print(step.next()) >>> print(six.next(step))
A woke up with A A woke up with A
returning g returning g
returning h returning h
returning i returning i
g g
>>> print("".join([step.next() for x in range(3)])) >>> print("".join([six.next(step) for x in range(3)]))
returning j returning j
returning k returning k
returning l returning l
returning m returning m
hij hij
>>> pausers[1].send("B") >>> pausers[1].send("B")
>>> print("".join([step.next() for x in range(4)])) >>> print("".join([six.next(step) for x in range(4)]))
B woke up with B B woke up with B
returning n returning n
returning o returning o
@@ -255,7 +257,7 @@ class Pool(object):
# but if we launched no coroutines with that queue as the destination, # but if we launched no coroutines with that queue as the destination,
# we could end up waiting a very long time. # we could end up waiting a very long time.
try: try:
index, args = tuples.next() index, args = six.next(tuples)
except StopIteration: except StopIteration:
return return
# From this point forward, 'args' is the current arguments tuple and # From this point forward, 'args' is the current arguments tuple and
@@ -297,7 +299,7 @@ class Pool(object):
# which to send() the result. # which to send() the result.
self._execute(q, function, args, {}) self._execute(q, function, args, {})
# We've consumed that args tuple, advance to next. # We've consumed that args tuple, advance to next.
index, args = tuples.next() index, args = six.next(tuples)
# Okay, we've filled up the pool again, yield a result -- which # Okay, we've filled up the pool again, yield a result -- which
# will probably wait for a coroutine to complete. Although we do # will probably wait for a coroutine to complete. Although we do
# have q.ready(), so we could iterate without waiting, we avoid # have q.ready(), so we could iterate without waiting, we avoid

48
eventlet/processes.py
View File

@@ -1,23 +1,22 @@
import warnings import warnings
warnings.warn("eventlet.processes is deprecated in favor of " warnings.warn("eventlet.processes is deprecated in favor of "
"eventlet.green.subprocess, which is API-compatible with the standard " "eventlet.green.subprocess, which is API-compatible with the standard "
" library subprocess module.", " library subprocess module.",
DeprecationWarning, stacklevel=2) DeprecationWarning, stacklevel=2)
import errno import errno
import os import os
import popen2
import signal import signal
from eventlet import api import eventlet
from eventlet import pools from eventlet import greenio, pools
from eventlet import greenio from eventlet.green import subprocess
class DeadProcess(RuntimeError): class DeadProcess(RuntimeError):
pass pass
def cooperative_wait(pobj, check_interval=0.01): def cooperative_wait(pobj, check_interval=0.01):
""" Waits for a child process to exit, returning the status """ Waits for a child process to exit, returning the status
code. code.
@@ -34,7 +33,7 @@ def cooperative_wait(pobj, check_interval=0.01):
status = pobj.poll() status = pobj.poll()
if status >= 0: if status >= 0:
return status return status
api.sleep(check_interval) eventlet.sleep(check_interval)
except OSError as e: except OSError as e:
if e.errno == errno.ECHILD: if e.errno == errno.ECHILD:
# no child process, this happens if the child process # no child process, this happens if the child process
@@ -48,7 +47,8 @@ def cooperative_wait(pobj, check_interval=0.01):
class Process(object): class Process(object):
"""Construct Process objects, then call read, and write on them.""" """Construct Process objects, then call read, and write on them."""
process_number = 0 process_number = 0
def __init__(self, command, args, dead_callback=lambda:None):
def __init__(self, command, args, dead_callback=None):
self.process_number = self.process_number + 1 self.process_number = self.process_number + 1
Process.process_number = self.process_number Process.process_number = self.process_number
self.command = command self.command = command
@@ -59,14 +59,20 @@ class Process(object):
def run(self): def run(self):
self.dead = False self.dead = False
self.started = False self.started = False
self.popen4 = None self.proc = None
## We use popen4 so that read() will read from either stdout or stderr args = [self.command]
self.popen4 = popen2.Popen4([self.command] + self.args) args.extend(self.args)
child_stdout_stderr = self.popen4.fromchild self.proc = subprocess.Popen(
child_stdin = self.popen4.tochild args=args,
self.child_stdout_stderr = greenio.GreenPipe(child_stdout_stderr, child_stdout_stderr.mode, 0) shell=False,
self.child_stdin = greenio.GreenPipe(child_stdin, child_stdin.mode, 0) stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
close_fds=True,
)
self.child_stdout_stderr = self.proc.stdout
self.child_stdin = self.proc.stdin
self.sendall = self.child_stdin.write self.sendall = self.child_stdin.write
self.send = self.child_stdin.write self.send = self.child_stdin.write
@@ -75,7 +81,7 @@ class Process(object):
self._read_first_result = False self._read_first_result = False
def wait(self): def wait(self):
return cooperative_wait(self.popen4) return cooperative_wait(self.proc)
def dead_callback(self): def dead_callback(self):
self.wait() self.wait()
@@ -128,13 +134,13 @@ class Process(object):
self.child_stdin.close() self.child_stdin.close()
def kill(self, sig=None): def kill(self, sig=None):
if sig == None: if sig is None:
sig = signal.SIGTERM sig = signal.SIGTERM
pid = self.getpid() pid = self.getpid()
os.kill(pid, sig) os.kill(pid, sig)
def getpid(self): def getpid(self):
return self.popen4.pid return self.proc.pid
class ProcessPool(pools.Pool): class ProcessPool(pools.Pool):
@@ -157,7 +163,7 @@ class ProcessPool(pools.Pool):
def put(self, item): def put(self, item):
if not item.dead: if not item.dead:
if item.popen4.poll() != -1: if item.proc.poll() != -1:
item.dead_callback() item.dead_callback()
else: else:
pools.Pool.put(self, item) pools.Pool.put(self, item)

62
eventlet/tpool.py
View File

@@ -16,6 +16,7 @@
import imp import imp
import os import os
import sys import sys
import traceback
from eventlet import event, greenio, greenthread, patcher, timeout from eventlet import event, greenio, greenthread, patcher, timeout
from eventlet.support import six from eventlet.support import six
@@ -32,12 +33,13 @@ Empty = Queue_module.Empty
__all__ = ['execute', 'Proxy', 'killall'] __all__ = ['execute', 'Proxy', 'killall']
QUIET=True QUIET = True
_rfile = _wfile = None _rfile = _wfile = None
_bytetosend = ' '.encode() _bytetosend = ' '.encode()
def _signal_t2e(): def _signal_t2e():
_wfile.write(_bytetosend) _wfile.write(_bytetosend)
_wfile.flush() _wfile.flush()
@@ -45,9 +47,10 @@ def _signal_t2e():
_reqq = None _reqq = None
_rspq = None _rspq = None
def tpool_trampoline(): def tpool_trampoline():
global _rspq global _rspq
while(True): while True:
try: try:
_c = _rfile.read(1) _c = _rfile.read(1)
assert _c assert _c
@@ -55,7 +58,7 @@ def tpool_trampoline():
break # will be raised when pipe is closed break # will be raised when pipe is closed
while not _rspq.empty(): while not _rspq.empty():
try: try:
(e,rv) = _rspq.get(block=False) (e, rv) = _rspq.get(block=False)
e.send(rv) e.send(rv)
e = rv = None e = rv = None
except Empty: except Empty:
@@ -65,31 +68,32 @@ def tpool_trampoline():
SYS_EXCS = (KeyboardInterrupt, SystemExit) SYS_EXCS = (KeyboardInterrupt, SystemExit)
EXC_CLASSES = (Exception, timeout.Timeout) EXC_CLASSES = (Exception, timeout.Timeout)
def tworker(): def tworker():
global _rspq global _rspq
while(True): while(True):
try: try:
msg = _reqq.get() msg = _reqq.get()
except AttributeError: except AttributeError:
return # can't get anything off of a dud queue return # can't get anything off of a dud queue
if msg is None: if msg is None:
return return
(e,meth,args,kwargs) = msg (e, meth, args, kwargs) = msg
rv = None rv = None
try: try:
rv = meth(*args,**kwargs) rv = meth(*args, **kwargs)
except SYS_EXCS: except SYS_EXCS:
raise raise
except EXC_CLASSES: except EXC_CLASSES:
rv = sys.exc_info() rv = sys.exc_info()
# test_leakage_from_tracebacks verifies that the use of # test_leakage_from_tracebacks verifies that the use of
# exc_info does not lead to memory leaks # exc_info does not lead to memory leaks
_rspq.put((e,rv)) _rspq.put((e, rv))
msg = meth = args = kwargs = e = rv = None msg = meth = args = kwargs = e = rv = None
_signal_t2e() _signal_t2e()
def execute(meth,*args, **kwargs): def execute(meth, *args, **kwargs):
""" """
Execute *meth* in a Python thread, blocking the current coroutine/ Execute *meth* in a Python thread, blocking the current coroutine/
greenthread until the method completes. greenthread until the method completes.
@@ -109,16 +113,15 @@ def execute(meth,*args, **kwargs):
return meth(*args, **kwargs) return meth(*args, **kwargs)
e = event.Event() e = event.Event()
_reqq.put((e,meth,args,kwargs)) _reqq.put((e, meth, args, kwargs))
rv = e.wait() rv = e.wait()
if isinstance(rv,tuple) \ if isinstance(rv, tuple) \
and len(rv) == 3 \ and len(rv) == 3 \
and isinstance(rv[1],EXC_CLASSES): and isinstance(rv[1], EXC_CLASSES):
import traceback (c, e, tb) = rv
(c,e,tb) = rv
if not QUIET: if not QUIET:
traceback.print_exception(c,e,tb) traceback.print_exception(c, e, tb)
traceback.print_stack() traceback.print_stack()
six.reraise(c, e, tb) six.reraise(c, e, tb)
return rv return rv
@@ -136,15 +139,16 @@ def proxy_call(autowrap, f, *args, **kwargs):
that don't need to be called in a separate thread, but which return objects that don't need to be called in a separate thread, but which return objects
that should be Proxy wrapped. that should be Proxy wrapped.
""" """
if kwargs.pop('nonblocking',False): if kwargs.pop('nonblocking', False):
rv = f(*args, **kwargs) rv = f(*args, **kwargs)
else: else:
rv = execute(f,*args,**kwargs) rv = execute(f, *args, **kwargs)
if isinstance(rv, autowrap): if isinstance(rv, autowrap):
return Proxy(rv, autowrap) return Proxy(rv, autowrap)
else: else:
return rv return rv
class Proxy(object): class Proxy(object):
""" """
a simple proxy-wrapper of any object that comes with a a simple proxy-wrapper of any object that comes with a
@@ -166,7 +170,7 @@ class Proxy(object):
of strings, which represent the names of attributes that should be of strings, which represent the names of attributes that should be
wrapped in Proxy objects when accessed. wrapped in Proxy objects when accessed.
""" """
def __init__(self, obj,autowrap=(), autowrap_names=()): def __init__(self, obj, autowrap=(), autowrap_names=()):
self._obj = obj self._obj = obj
self._autowrap = autowrap self._autowrap = autowrap
self._autowrap_names = autowrap_names self._autowrap_names = autowrap_names
@@ -177,6 +181,7 @@ class Proxy(object):
if isinstance(f, self._autowrap) or attr_name in self._autowrap_names: if isinstance(f, self._autowrap) or attr_name in self._autowrap_names:
return Proxy(f, self._autowrap) return Proxy(f, self._autowrap)
return f return f
def doit(*args, **kwargs): def doit(*args, **kwargs):
result = proxy_call(self._autowrap, f, *args, **kwargs) result = proxy_call(self._autowrap, f, *args, **kwargs)
if attr_name in self._autowrap_names and not isinstance(result, Proxy): if attr_name in self._autowrap_names and not isinstance(result, Proxy):
@@ -189,19 +194,25 @@ class Proxy(object):
# explicitly # explicitly
def __getitem__(self, key): def __getitem__(self, key):
return proxy_call(self._autowrap, self._obj.__getitem__, key) return proxy_call(self._autowrap, self._obj.__getitem__, key)
def __setitem__(self, key, value): def __setitem__(self, key, value):
return proxy_call(self._autowrap, self._obj.__setitem__, key, value) return proxy_call(self._autowrap, self._obj.__setitem__, key, value)
def __deepcopy__(self, memo=None): def __deepcopy__(self, memo=None):
return proxy_call(self._autowrap, self._obj.__deepcopy__, memo) return proxy_call(self._autowrap, self._obj.__deepcopy__, memo)
def __copy__(self, memo=None): def __copy__(self, memo=None):
return proxy_call(self._autowrap, self._obj.__copy__, memo) return proxy_call(self._autowrap, self._obj.__copy__, memo)
def __call__(self, *a, **kw): def __call__(self, *a, **kw):
if '__call__' in self._autowrap_names: if '__call__' in self._autowrap_names:
return Proxy(proxy_call(self._autowrap, self._obj, *a, **kw)) return Proxy(proxy_call(self._autowrap, self._obj, *a, **kw))
else: else:
return proxy_call(self._autowrap, self._obj, *a, **kw) return proxy_call(self._autowrap, self._obj, *a, **kw)
def __enter__(self): def __enter__(self):
return proxy_call(self._autowrap, self._obj.__enter__) return proxy_call(self._autowrap, self._obj.__enter__)
def __exit__(self, *exc): def __exit__(self, *exc):
return proxy_call(self._autowrap, self._obj.__exit__, *exc) return proxy_call(self._autowrap, self._obj.__exit__, *exc)
@@ -210,31 +221,43 @@ class Proxy(object):
# wrapped object in such a way that they would block # wrapped object in such a way that they would block
def __eq__(self, rhs): def __eq__(self, rhs):
return self._obj == rhs return self._obj == rhs
def __hash__(self): def __hash__(self):
return self._obj.__hash__() return self._obj.__hash__()
def __repr__(self): def __repr__(self):
return self._obj.__repr__() return self._obj.__repr__()
def __str__(self): def __str__(self):
return self._obj.__str__() return self._obj.__str__()
def __len__(self): def __len__(self):
return len(self._obj) return len(self._obj)
def __nonzero__(self): def __nonzero__(self):
return bool(self._obj) return bool(self._obj)
# Python3
__bool__ = __nonzero__ __bool__ = __nonzero__
def __iter__(self): def __iter__(self):
it = iter(self._obj) it = iter(self._obj)
if it == self._obj: if it == self._obj:
return self return self
else: else:
return Proxy(it) return Proxy(it)
def next(self): def next(self):
return proxy_call(self._autowrap, self._obj.next) return proxy_call(self._autowrap, self._obj.next)
# Python3
__next__ = next
_nthreads = int(os.environ.get('EVENTLET_THREADPOOL_SIZE', 20)) _nthreads = int(os.environ.get('EVENTLET_THREADPOOL_SIZE', 20))
_threads = [] _threads = []
_coro = None _coro = None
_setup_already = False _setup_already = False
def setup(): def setup():
global _rfile, _wfile, _threads, _coro, _setup_already, _rspq, _reqq global _rfile, _wfile, _threads, _coro, _setup_already, _rspq, _reqq
if _setup_already: if _setup_already:
@@ -257,7 +280,7 @@ def setup():
csock.connect(('localhost', sock.getsockname()[1])) csock.connect(('localhost', sock.getsockname()[1]))
nsock, addr = sock.accept() nsock, addr = sock.accept()
_rfile = greenio.GreenSocket(csock).makefile('rb', 0) _rfile = greenio.GreenSocket(csock).makefile('rb', 0)
_wfile = nsock.makefile('wb',0) _wfile = nsock.makefile('wb', 0)
_reqq = Queue(maxsize=-1) _reqq = Queue(maxsize=-1)
_rspq = Queue(maxsize=-1) _rspq = Queue(maxsize=-1)
@@ -295,6 +318,7 @@ def killall():
_rspq = None _rspq = None
_setup_already = False _setup_already = False
def set_num_threads(nthreads): def set_num_threads(nthreads):
global _nthreads global _nthreads
_nthreads = nthreads _nthreads = nthreads

5
tests/__init__.py
View File

@@ -22,11 +22,6 @@ from eventlet import tpool
main = unittest.main main = unittest.main
def s2b(s):
"""portable way to convert string to bytes. In 3.x socket.send and recv require bytes"""
return s.encode()
def skipped(func): def skipped(func):
""" Decorator that marks a function as skipped. Uses nose's SkipTest exception """ Decorator that marks a function as skipped. Uses nose's SkipTest exception
if installed. Without nose, this will count skipped tests as passing tests.""" if installed. Without nose, this will count skipped tests as passing tests."""

30
tests/api_test.py
View File

@@ -4,12 +4,12 @@ from unittest import TestCase, main
import warnings import warnings
import eventlet import eventlet
from eventlet import greenio, util, hubs, greenthread, spawn
from tests import skip_if_no_ssl
warnings.simplefilter('ignore', DeprecationWarning) warnings.simplefilter('ignore', DeprecationWarning)
from eventlet import api from eventlet import api
warnings.simplefilter('default', DeprecationWarning) warnings.simplefilter('default', DeprecationWarning)
from eventlet import greenio, util, hubs, greenthread, spawn
from tests import skip_if_no_ssl
def check_hub(): def check_hub():
@@ -57,7 +57,7 @@ class TestApi(TestCase):
client.close() client.close()
assert fd.readline() == b'hello\n' assert fd.readline() == b'hello\n'
assert fd.read() == '' assert fd.read() == b''
fd.close() fd.close()
check_hub() check_hub()
@@ -85,7 +85,7 @@ class TestApi(TestCase):
assert fd.readline() == b'hello\r\n' assert fd.readline() == b'hello\r\n'
try: try:
self.assertEquals('', fd.read(10)) self.assertEqual(b'', fd.read(10))
except greenio.SSL.ZeroReturnError: except greenio.SSL.ZeroReturnError:
# if it's a GreenSSL object it'll do this # if it's a GreenSSL object it'll do this
pass pass
@@ -147,9 +147,7 @@ class TestApi(TestCase):
def test_named(self): def test_named(self):
named_foo = api.named('tests.api_test.Foo') named_foo = api.named('tests.api_test.Foo')
self.assertEquals( self.assertEqual(named_foo.__name__, "Foo")
named_foo.__name__,
"Foo")
def test_naming_missing_class(self): def test_naming_missing_class(self):
self.assertRaises( self.assertRaises(
@@ -171,19 +169,25 @@ class TestApi(TestCase):
# thus after the function's done, the control will go to the parent # thus after the function's done, the control will go to the parent
api.sleep(0) api.sleep(0)
state.append('finished') state.append('finished')
g = api.spawn(test) g = api.spawn(test)
api.sleep(DELAY/2) api.sleep(DELAY / 2)
self.assertEquals(state, ['start']) self.assertEqual(state, ['start'])
api.kill(g) api.kill(g)
# will not get there, unless switching is explicitly scheduled by kill # will not get there, unless switching is explicitly scheduled by kill
self.assertEquals(state, ['start', 'except']) self.assertEqual(state, ['start', 'except'])
api.sleep(DELAY) api.sleep(DELAY)
self.assertEquals(state, ['start', 'except', 'finished']) self.assertEqual(state, ['start', 'except', 'finished'])
def test_nested_with_timeout(self): def test_nested_with_timeout(self):
def func(): def func():
return api.with_timeout(0.2, api.sleep, 2, timeout_value=1) return api.with_timeout(0.2, api.sleep, 2, timeout_value=1)
self.assertRaises(api.TimeoutError, api.with_timeout, 0.1, func)
try:
api.with_timeout(0.1, func)
self.fail(u'Expected api.TimeoutError')
except api.TimeoutError:
pass
class Foo(object): class Foo(object):

6
tests/backdoor_test.py
View File

@@ -18,11 +18,11 @@ class BackdoorTest(LimitedTestCase):
f.readline() # build info f.readline() # build info
f.readline() # help info f.readline() # help info
self.assert_(b'InteractiveConsole' in f.readline()) self.assert_(b'InteractiveConsole' in f.readline())
self.assertEquals(b'>>> ', f.read(4)) self.assertEqual(b'>>> ', f.read(4))
f.write(b'print("hi")\n') f.write(b'print("hi")\n')
f.flush() f.flush()
self.assertEquals(b'hi\n', f.readline()) self.assertEqual(b'hi\n', f.readline())
self.assertEquals(b'>>> ', f.read(4)) self.assertEqual(b'>>> ', f.read(4))
f.close() f.close()
client.close() client.close()
serv.kill() serv.kill()

67
tests/convenience_test.py
View File

@@ -1,10 +1,10 @@
import os import os
import eventlet import eventlet
from eventlet import event from eventlet import debug, event
from eventlet.green import socket from eventlet.green import socket
from eventlet.support import six from eventlet.support import six
from tests import LimitedTestCase, s2b, skip_if_no_ssl from tests import LimitedTestCase, skip_if_no_ssl
certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt') certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
@@ -14,43 +14,40 @@ private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')
class TestServe(LimitedTestCase): class TestServe(LimitedTestCase):
def setUp(self): def setUp(self):
super(TestServe, self).setUp() super(TestServe, self).setUp()
from eventlet import debug
debug.hub_exceptions(False) debug.hub_exceptions(False)
def tearDown(self): def tearDown(self):
super(TestServe, self).tearDown() super(TestServe, self).tearDown()
from eventlet import debug
debug.hub_exceptions(True) debug.hub_exceptions(True)
def test_exiting_server(self): def test_exiting_server(self):
# tests that the server closes the client sock on handle() exit # tests that the server closes the client sock on handle() exit
def closer(sock,addr): def closer(sock, addr):
pass pass
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, closer) gt = eventlet.spawn(eventlet.serve, l, closer)
client = eventlet.connect(('localhost', l.getsockname()[1])) client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall(s2b('a')) client.sendall(b'a')
self.assertFalse(client.recv(100)) self.assertFalse(client.recv(100))
gt.kill() gt.kill()
def test_excepting_server(self): def test_excepting_server(self):
# tests that the server closes the client sock on handle() exception # tests that the server closes the client sock on handle() exception
def crasher(sock,addr): def crasher(sock, addr):
sock.recv(1024) sock.recv(1024)
0//0 0//0
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, crasher) gt = eventlet.spawn(eventlet.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1])) client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall(s2b('a')) client.sendall(b'a')
self.assertRaises(ZeroDivisionError, gt.wait) self.assertRaises(ZeroDivisionError, gt.wait)
self.assertFalse(client.recv(100)) self.assertFalse(client.recv(100))
def test_excepting_server_already_closed(self): def test_excepting_server_already_closed(self):
# same as above but with explicit clsoe before crash # same as above but with explicit clsoe before crash
def crasher(sock,addr): def crasher(sock, addr):
sock.recv(1024) sock.recv(1024)
sock.close() sock.close()
0//0 0//0
@@ -58,14 +55,15 @@ class TestServe(LimitedTestCase):
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, crasher) gt = eventlet.spawn(eventlet.serve, l, crasher)
client = eventlet.connect(('localhost', l.getsockname()[1])) client = eventlet.connect(('localhost', l.getsockname()[1]))
client.sendall(s2b('a')) client.sendall(b'a')
self.assertRaises(ZeroDivisionError, gt.wait) self.assertRaises(ZeroDivisionError, gt.wait)
self.assertFalse(client.recv(100)) self.assertFalse(client.recv(100))
def test_called_for_each_connection(self): def test_called_for_each_connection(self):
hits = [0] hits = [0]
def counter(sock, addr): def counter(sock, addr):
hits[0]+=1 hits[0] += 1
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, counter) gt = eventlet.spawn(eventlet.serve, l, counter)
for i in six.moves.range(100): for i in six.moves.range(100):
@@ -76,8 +74,9 @@ class TestServe(LimitedTestCase):
def test_blocking(self): def test_blocking(self):
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
x = eventlet.with_timeout(0.01, x = eventlet.with_timeout(
eventlet.serve, l, lambda c,a: None, 0.01,
eventlet.serve, l, lambda c, a: None,
timeout_value="timeout") timeout_value="timeout")
self.assertEqual(x, "timeout") self.assertEqual(x, "timeout")
@@ -86,50 +85,56 @@ class TestServe(LimitedTestCase):
raise eventlet.StopServe() raise eventlet.StopServe()
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
# connect to trigger a call to stopit # connect to trigger a call to stopit
gt = eventlet.spawn(eventlet.connect, gt = eventlet.spawn(eventlet.connect, ('localhost', l.getsockname()[1]))
('localhost', l.getsockname()[1]))
eventlet.serve(l, stopit) eventlet.serve(l, stopit)
gt.wait() gt.wait()
def test_concurrency(self): def test_concurrency(self):
evt = event.Event() evt = event.Event()
def waiter(sock, addr): def waiter(sock, addr):
sock.sendall(s2b('hi')) sock.sendall(b'hi')
evt.wait() evt.wait()
l = eventlet.listen(('localhost', 0)) l = eventlet.listen(('localhost', 0))
gt = eventlet.spawn(eventlet.serve, l, waiter, 5) eventlet.spawn(eventlet.serve, l, waiter, 5)
def test_client(): def test_client():
c = eventlet.connect(('localhost', l.getsockname()[1])) c = eventlet.connect(('localhost', l.getsockname()[1]))
# verify the client is connected by getting data # verify the client is connected by getting data
self.assertEquals(s2b('hi'), c.recv(2)) self.assertEqual(b'hi', c.recv(2))
return c return c
clients = [test_client() for i in range(5)] [test_client() for i in range(5)]
# very next client should not get anything # very next client should not get anything
x = eventlet.with_timeout(0.01, x = eventlet.with_timeout(
0.01,
test_client, test_client,
timeout_value="timed out") timeout_value="timed out")
self.assertEquals(x, "timed out") self.assertEqual(x, "timed out")
@skip_if_no_ssl @skip_if_no_ssl
def test_wrap_ssl(self): def test_wrap_ssl(self):
server = eventlet.wrap_ssl(eventlet.listen(('localhost', 0)), server = eventlet.wrap_ssl(
certfile=certificate_file, eventlet.listen(('localhost', 0)),
keyfile=private_key_file, server_side=True) certfile=certificate_file, keyfile=private_key_file,
server_side=True)
port = server.getsockname()[1] port = server.getsockname()[1]
def handle(sock,addr):
def handle(sock, addr):
sock.sendall(sock.recv(1024)) sock.sendall(sock.recv(1024))
raise eventlet.StopServe() raise eventlet.StopServe()
eventlet.spawn(eventlet.serve, server, handle) eventlet.spawn(eventlet.serve, server, handle)
client = eventlet.wrap_ssl(eventlet.connect(('localhost', port))) client = eventlet.wrap_ssl(eventlet.connect(('localhost', port)))
client.sendall("echo") client.sendall("echo")
self.assertEquals("echo", client.recv(1024)) self.assertEqual("echo", client.recv(1024))
def test_socket_reuse(self): def test_socket_reuse(self):
lsock1 = eventlet.listen(('localhost',0)) lsock1 = eventlet.listen(('localhost', 0))
port = lsock1.getsockname()[1] port = lsock1.getsockname()[1]
def same_socket(): def same_socket():
return eventlet.listen(('localhost',port)) return eventlet.listen(('localhost', port))
self.assertRaises(socket.error,same_socket)
self.assertRaises(socket.error, same_socket)
lsock1.close() lsock1.close()
assert same_socket() assert same_socket()

60
tests/db_pool_test.py
View File

@@ -265,19 +265,19 @@ class DBConnectionPool(DBTester):
conn = self.pool._unwrap_connection(self.connection) conn = self.pool._unwrap_connection(self.connection)
self.assert_(not isinstance(conn, db_pool.GenericConnectionWrapper)) self.assert_(not isinstance(conn, db_pool.GenericConnectionWrapper))
self.assertEquals(None, self.pool._unwrap_connection(None)) self.assertEqual(None, self.pool._unwrap_connection(None))
self.assertEquals(None, self.pool._unwrap_connection(1)) self.assertEqual(None, self.pool._unwrap_connection(1))
# testing duck typing here -- as long as the connection has a # testing duck typing here -- as long as the connection has a
# _base attribute, it should be unwrappable # _base attribute, it should be unwrappable
x = Mock() x = Mock()
x._base = 'hi' x._base = 'hi'
self.assertEquals('hi', self.pool._unwrap_connection(x)) self.assertEqual('hi', self.pool._unwrap_connection(x))
conn.close() conn.close()
def test_safe_close(self): def test_safe_close(self):
self.pool._safe_close(self.connection, quiet=True) self.pool._safe_close(self.connection, quiet=True)
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
self.pool._safe_close(None) self.pool._safe_close(None)
self.pool._safe_close(1) self.pool._safe_close(1)
@@ -301,7 +301,7 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=2, max_idle=0) self.pool = self.create_pool(max_size=2, max_idle=0)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
def test_zero_max_age(self): def test_zero_max_age(self):
self.pool.put(self.connection) self.pool.put(self.connection)
@@ -309,7 +309,7 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=2, max_age=0) self.pool = self.create_pool(max_size=2, max_age=0)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
@skipped @skipped
def test_max_idle(self): def test_max_idle(self):
@@ -320,19 +320,19 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=2, max_idle=0.02) self.pool = self.create_pool(max_size=2, max_idle=0.02)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.01) # not long enough to trigger the idle timeout eventlet.sleep(0.01) # not long enough to trigger the idle timeout
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.01) # idle timeout should have fired but done nothing eventlet.sleep(0.01) # idle timeout should have fired but done nothing
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.03) # long enough to trigger idle timeout for real eventlet.sleep(0.03) # long enough to trigger idle timeout for real
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
@skipped @skipped
def test_max_idle_many(self): def test_max_idle_many(self):
@@ -344,11 +344,11 @@ class DBConnectionPool(DBTester):
self.connection, conn2 = self.pool.get(), self.pool.get() self.connection, conn2 = self.pool.get(), self.pool.get()
self.connection.close() self.connection.close()
eventlet.sleep(0.01) eventlet.sleep(0.01)
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
conn2.close() conn2.close()
self.assertEquals(len(self.pool.free_items), 2) self.assertEqual(len(self.pool.free_items), 2)
eventlet.sleep(0.02) # trigger cleanup of conn1 but not conn2 eventlet.sleep(0.02) # trigger cleanup of conn1 but not conn2
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
@skipped @skipped
def test_max_age(self): def test_max_age(self):
@@ -359,14 +359,14 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=2, max_age=0.05) self.pool = self.create_pool(max_size=2, max_age=0.05)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.01) # not long enough to trigger the age timeout eventlet.sleep(0.01) # not long enough to trigger the age timeout
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
self.connection = self.pool.get() self.connection = self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.05) # long enough to trigger age timeout eventlet.sleep(0.05) # long enough to trigger age timeout
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
@skipped @skipped
def test_max_age_many(self): def test_max_age_many(self):
@@ -377,13 +377,13 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=2, max_age=0.15) self.pool = self.create_pool(max_size=2, max_age=0.15)
self.connection, conn2 = self.pool.get(), self.pool.get() self.connection, conn2 = self.pool.get(), self.pool.get()
self.connection.close() self.connection.close()
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0) # not long enough to trigger the age timeout eventlet.sleep(0) # not long enough to trigger the age timeout
self.assertEquals(len(self.pool.free_items), 1) self.assertEqual(len(self.pool.free_items), 1)
eventlet.sleep(0.2) # long enough to trigger age timeout eventlet.sleep(0.2) # long enough to trigger age timeout
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
conn2.close() # should not be added to the free items conn2.close() # should not be added to the free items
self.assertEquals(len(self.pool.free_items), 0) self.assertEqual(len(self.pool.free_items), 0)
def test_waiters_get_woken(self): def test_waiters_get_woken(self):
# verify that when there's someone waiting on an empty pool # verify that when there's someone waiting on an empty pool
@@ -394,8 +394,8 @@ class DBConnectionPool(DBTester):
self.pool = self.create_pool(max_size=1, max_age=0) self.pool = self.create_pool(max_size=1, max_age=0)
self.connection = self.pool.get() self.connection = self.pool.get()
self.assertEquals(self.pool.free(), 0) self.assertEqual(self.pool.free(), 0)
self.assertEquals(self.pool.waiting(), 0) self.assertEqual(self.pool.waiting(), 0)
e = event.Event() e = event.Event()
def retrieve(pool, ev): def retrieve(pool, ev):
c = pool.get() c = pool.get()
@@ -403,14 +403,14 @@ class DBConnectionPool(DBTester):
eventlet.spawn(retrieve, self.pool, e) eventlet.spawn(retrieve, self.pool, e)
eventlet.sleep(0) # these two sleeps should advance the retrieve eventlet.sleep(0) # these two sleeps should advance the retrieve
eventlet.sleep(0) # coroutine until it's waiting in get() eventlet.sleep(0) # coroutine until it's waiting in get()
self.assertEquals(self.pool.free(), 0) self.assertEqual(self.pool.free(), 0)
self.assertEquals(self.pool.waiting(), 1) self.assertEqual(self.pool.waiting(), 1)
self.pool.put(self.connection) self.pool.put(self.connection)
timer = eventlet.Timeout(1) timer = eventlet.Timeout(1)
conn = e.wait() conn = e.wait()
timer.cancel() timer.cancel()
self.assertEquals(self.pool.free(), 0) self.assertEqual(self.pool.free(), 0)
self.assertEquals(self.pool.waiting(), 0) self.assertEqual(self.pool.waiting(), 0)
self.pool.put(conn) self.pool.put(conn)
@skipped @skipped
@@ -440,7 +440,7 @@ class DBConnectionPool(DBTester):
# not lose any connections # not lose any connections
self.pool = self.create_pool(max_size=1, module=RaisingDBModule()) self.pool = self.create_pool(max_size=1, module=RaisingDBModule())
self.assertRaises(RuntimeError, self.pool.get) self.assertRaises(RuntimeError, self.pool.get)
self.assertEquals(self.pool.free(), 1) self.assertEqual(self.pool.free(), 1)
class DummyConnection(object): class DummyConnection(object):

6
tests/debug_test.py
View File

@@ -3,7 +3,7 @@ from unittest import TestCase
from eventlet import debug from eventlet import debug
from eventlet.support import six from eventlet.support import six
from tests import LimitedTestCase, main, s2b from tests import LimitedTestCase, main
import eventlet import eventlet
@@ -113,7 +113,7 @@ class TestDebug(LimitedTestCase):
try: try:
gt = eventlet.spawn(hurl, client_2) gt = eventlet.spawn(hurl, client_2)
eventlet.sleep(0) eventlet.sleep(0)
client.send(s2b(' ')) client.send(b' ')
eventlet.sleep(0) eventlet.sleep(0)
# allow the "hurl" greenlet to trigger the KeyError # allow the "hurl" greenlet to trigger the KeyError
# not sure why the extra context switch is needed # not sure why the extra context switch is needed
@@ -124,7 +124,7 @@ class TestDebug(LimitedTestCase):
debug.hub_exceptions(False) debug.hub_exceptions(False)
# look for the KeyError exception in the traceback # look for the KeyError exception in the traceback
self.assert_('KeyError: 1' in fake.getvalue(), self.assert_('KeyError: 1' in fake.getvalue(),
"Traceback not in:\n" + fake.getvalue()) "Traceback not in:\n" + fake.getvalue())
if __name__ == "__main__": if __name__ == "__main__":
main() main()

137
tests/greenio_test.py
View File

@@ -1,18 +1,26 @@
import socket as _orig_sock
from tests import LimitedTestCase, skip_with_pyevent, main, skipped, s2b, skip_if, skip_on_windows
from eventlet import event, greenio, debug
from eventlet.hubs import get_hub
from eventlet.green import select, socket, time, ssl
from eventlet.support import get_errno
import array import array
import errno import errno
import eventlet import eventlet
import gc
import fcntl import fcntl
import gc
import os import os
import shutil
import socket as _orig_sock
import sys import sys
import tempfile, shutil import tempfile
from eventlet import event, greenio, debug
from eventlet.hubs import get_hub
from eventlet.green import select, socket, time, ssl
from eventlet.support import get_errno, six
from tests import (
LimitedTestCase, main,
skip_with_pyevent, skipped, skip_if, skip_on_windows,
)
if six.PY3:
buffer = memoryview
def bufsized(sock, size=1): def bufsized(sock, size=1):
@@ -35,17 +43,17 @@ def min_buf_size():
def using_epoll_hub(_f): def using_epoll_hub(_f):
try: try:
return 'epolls' in type(get_hub()).__module__ return 'epolls' in type(get_hub()).__module__
except Exception: except Exception:
return False return False
def using_kqueue_hub(_f): def using_kqueue_hub(_f):
try: try:
return 'kqueue' in type(get_hub()).__module__ return 'kqueue' in type(get_hub()).__module__
except Exception: except Exception:
return False return False
class TestGreenSocket(LimitedTestCase): class TestGreenSocket(LimitedTestCase):
@@ -95,7 +103,7 @@ class TestGreenSocket(LimitedTestCase):
gs = greenio.GreenSocket(s) gs = greenio.GreenSocket(s)
e = gs.connect_ex(('192.0.2.1', 80)) e = gs.connect_ex(('192.0.2.1', 80))
if not e in (errno.EHOSTUNREACH, errno.ENETUNREACH): if not e in (errno.EHOSTUNREACH, errno.ENETUNREACH):
self.assertEquals(e, errno.EAGAIN) self.assertEqual(e, errno.EAGAIN)
def test_recv_timeout(self): def test_recv_timeout(self):
listener = greenio.GreenSocket(socket.socket()) listener = greenio.GreenSocket(socket.socket())
@@ -142,7 +150,7 @@ class TestGreenSocket(LimitedTestCase):
self.assertEqual(e.args[0], 'timed out') self.assertEqual(e.args[0], 'timed out')
def test_recvfrom_into_timeout(self): def test_recvfrom_into_timeout(self):
buf = buffer(array.array('B')) buf = array.array('B')
gs = greenio.GreenSocket( gs = greenio.GreenSocket(
socket.socket(socket.AF_INET, socket.SOCK_DGRAM)) socket.socket(socket.AF_INET, socket.SOCK_DGRAM))
@@ -157,7 +165,7 @@ class TestGreenSocket(LimitedTestCase):
self.assertEqual(e.args[0], 'timed out') self.assertEqual(e.args[0], 'timed out')
def test_recv_into_timeout(self): def test_recv_into_timeout(self):
buf = buffer(array.array('B')) buf = array.array('B')
listener = greenio.GreenSocket(socket.socket()) listener = greenio.GreenSocket(socket.socket())
listener.bind(('', 0)) listener.bind(('', 0))
@@ -209,7 +217,7 @@ class TestGreenSocket(LimitedTestCase):
client.connect(addr) client.connect(addr)
try: try:
client.settimeout(0.00001) client.settimeout(0.00001)
msg = s2b("A") * 100000 # large enough number to overwhelm most buffers msg = b"A" * 100000 # large enough number to overwhelm most buffers
total_sent = 0 total_sent = 0
# want to exceed the size of the OS buffer so it'll block in a # want to exceed the size of the OS buffer so it'll block in a
@@ -245,7 +253,7 @@ class TestGreenSocket(LimitedTestCase):
client.connect(addr) client.connect(addr)
try: try:
msg = s2b("A") * (8 << 20) msg = b"A" * (8 << 20)
# want to exceed the size of the OS buffer so it'll block # want to exceed the size of the OS buffer so it'll block
client.sendall(msg) client.sendall(msg)
@@ -265,10 +273,10 @@ class TestGreenSocket(LimitedTestCase):
conn, addr = listener.accept() conn, addr = listener.accept()
fd = conn.makefile('w') fd = conn.makefile('w')
conn.close() conn.close()
fd.write('hello\n') fd.write(b'hello\n')
fd.close() fd.close()
self.assertWriteToClosedFileRaises(fd) self.assertWriteToClosedFileRaises(fd)
self.assertRaises(socket.error, conn.send, s2b('b')) self.assertRaises(socket.error, conn.send, b'b')
finally: finally:
listener.close() listener.close()
@@ -278,12 +286,12 @@ class TestGreenSocket(LimitedTestCase):
try: try:
conn, addr = listener.accept() conn, addr = listener.accept()
fd = conn.makefile('w') fd = conn.makefile('w')
fd.write('hello') fd.write(b'hello')
fd.close() fd.close()
conn.send(s2b('\n')) conn.send(b'\n')
conn.close() conn.close()
self.assertWriteToClosedFileRaises(fd) self.assertWriteToClosedFileRaises(fd)
self.assertRaises(socket.error, conn.send, s2b('b')) self.assertRaises(socket.error, conn.send, b'b')
finally: finally:
listener.close() listener.close()
@@ -292,8 +300,8 @@ class TestGreenSocket(LimitedTestCase):
client.connect(('127.0.0.1', server.getsockname()[1])) client.connect(('127.0.0.1', server.getsockname()[1]))
fd = client.makefile() fd = client.makefile()
client.close() client.close()
assert fd.readline() == 'hello\n' assert fd.readline() == b'hello\n'
assert fd.read() == '' assert fd.read() == b''
fd.close() fd.close()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -320,7 +328,7 @@ class TestGreenSocket(LimitedTestCase):
try: try:
conn, addr = listener.accept() conn, addr = listener.accept()
conn = conn.makefile('w') conn = conn.makefile('w')
conn.write('hello\n') conn.write(b'hello\n')
conn.close() conn.close()
gc.collect() gc.collect()
self.assertWriteToClosedFileRaises(conn) self.assertWriteToClosedFileRaises(conn)
@@ -336,13 +344,13 @@ class TestGreenSocket(LimitedTestCase):
client.connect(('127.0.0.1', server.getsockname()[1])) client.connect(('127.0.0.1', server.getsockname()[1]))
fd = client.makefile() fd = client.makefile()
client.close() client.close()
assert fd.read() == 'hello\n' assert fd.read() == b'hello\n'
assert fd.read() == '' assert fd.read() == b''
killer.wait() killer.wait()
def test_full_duplex(self): def test_full_duplex(self):
large_data = s2b('*') * 10 * min_buf_size() large_data = b'*' * 10 * min_buf_size()
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
listener.bind(('127.0.0.1', 0)) listener.bind(('127.0.0.1', 0))
@@ -356,7 +364,7 @@ class TestGreenSocket(LimitedTestCase):
result = sock.recv(len(large_data)) result = sock.recv(len(large_data))
while len(result) < len(large_data): while len(result) < len(large_data):
result += sock.recv(len(large_data)) result += sock.recv(len(large_data))
self.assertEquals(result, large_data) self.assertEqual(result, large_data)
def server(): def server():
(sock, addr) = listener.accept() (sock, addr) = listener.accept()
@@ -364,10 +372,10 @@ class TestGreenSocket(LimitedTestCase):
send_large_coro = eventlet.spawn(send_large, sock) send_large_coro = eventlet.spawn(send_large, sock)
eventlet.sleep(0) eventlet.sleep(0)
result = sock.recv(10) result = sock.recv(10)
expected = s2b('hello world') expected = b'hello world'
while len(result) < len(expected): while len(result) < len(expected):
result += sock.recv(10) result += sock.recv(10)
self.assertEquals(result, expected) self.assertEqual(result, expected)
send_large_coro.wait() send_large_coro.wait()
server_evt = eventlet.spawn(server) server_evt = eventlet.spawn(server)
@@ -376,7 +384,7 @@ class TestGreenSocket(LimitedTestCase):
bufsized(client) bufsized(client)
large_evt = eventlet.spawn(read_large, client) large_evt = eventlet.spawn(read_large, client)
eventlet.sleep(0) eventlet.sleep(0)
client.sendall(s2b('hello world')) client.sendall(b'hello world')
server_evt.wait() server_evt.wait()
large_evt.wait() large_evt.wait()
client.close() client.close()
@@ -393,8 +401,8 @@ class TestGreenSocket(LimitedTestCase):
def sender(listener): def sender(listener):
(sock, addr) = listener.accept() (sock, addr) = listener.accept()
sock = bufsized(sock, size=bufsize) sock = bufsized(sock, size=bufsize)
sock.sendall(s2b('x') * many_bytes) sock.sendall(b'x' * many_bytes)
sock.sendall(s2b('y') * second_bytes) sock.sendall(b'y' * second_bytes)
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
@@ -447,12 +455,13 @@ class TestGreenSocket(LimitedTestCase):
eventlet.sleep(0.02) eventlet.sleep(0.02)
wrap_wfile.write('hi') wrap_wfile.write('hi')
s2.close() s2.close()
evt.send('sent via event') evt.send(b'sent via event')
evt = event.Event() evt = event.Event()
eventlet.spawn(sender, evt) eventlet.spawn(sender, evt)
eventlet.sleep(0) # lets the socket enter accept mode, which # lets the socket enter accept mode, which
# is necessary for connect to succeed on windows # is necessary for connect to succeed on windows
eventlet.sleep(0)
try: try:
# try and get some data off of this pipe # try and get some data off of this pipe
# but bail before any is sent # but bail before any is sent
@@ -466,7 +475,7 @@ class TestGreenSocket(LimitedTestCase):
pass pass
result = evt.wait() result = evt.wait()
self.assertEquals(result, 'sent via event') self.assertEqual(result, b'sent via event')
server.close() server.close()
client.close() client.close()
@@ -476,7 +485,7 @@ class TestGreenSocket(LimitedTestCase):
def handle(sock, addr): def handle(sock, addr):
sock.recv(1) sock.recv(1)
sock.sendall("a") sock.sendall(b"a")
raise eventlet.StopServe() raise eventlet.StopServe()
listener = eventlet.listen(('127.0.0.1', 0)) listener = eventlet.listen(('127.0.0.1', 0))
@@ -489,7 +498,7 @@ class TestGreenSocket(LimitedTestCase):
a = eventlet.spawn(reader, s) a = eventlet.spawn(reader, s)
eventlet.sleep(0) eventlet.sleep(0)
self.assertRaises(RuntimeError, s.recv, 1) self.assertRaises(RuntimeError, s.recv, 1)
s.sendall('b') s.sendall(b'b')
a.wait() a.wait()
@skip_with_pyevent @skip_with_pyevent
@@ -500,7 +509,7 @@ class TestGreenSocket(LimitedTestCase):
sock = eventlet.connect(address) sock = eventlet.connect(address)
while True: while True:
try: try:
sock.sendall('hello world') sock.sendall(b'hello world')
except socket.error as e: except socket.error as e:
if get_errno(e) == errno.EPIPE: if get_errno(e) == errno.EPIPE:
return return
@@ -628,8 +637,8 @@ class TestGreenPipe(LimitedTestCase):
for i in range(5): for i in range(5):
line = rf.readline() line = rf.readline()
eventlet.sleep(0.01) eventlet.sleep(0.01)
self.assertEquals(line, one_line) self.assertEqual(line, one_line)
self.assertEquals(rf.readline(), '') self.assertEqual(rf.readline(), '')
def test_pipe_read(self): def test_pipe_read(self):
# ensure that 'readline' works properly on GreenPipes when data is not # ensure that 'readline' works properly on GreenPipes when data is not
@@ -654,10 +663,10 @@ class TestGreenPipe(LimitedTestCase):
eventlet.sleep(0) eventlet.sleep(0)
line = r.readline() line = r.readline()
self.assertEquals(line, 'line\n') self.assertEqual(line, 'line\n')
line = r.readline() line = r.readline()
self.assertEquals(line, 'line\r\n') self.assertEqual(line, 'line\r\n')
gt.wait() gt.wait()
@@ -678,26 +687,27 @@ class TestGreenPipe(LimitedTestCase):
for i in range(65): for i in range(65):
buf = r.read(1024) buf = r.read(1024)
expected = 1024 * chr(i) expected = 1024 * chr(i)
self.assertEquals(buf, expected, self.assertEqual(
buf, expected,
"expected=%r..%r, found=%r..%r iter=%d" "expected=%r..%r, found=%r..%r iter=%d"
% (expected[:4], expected[-4:], buf[:4], buf[-4:], i)) % (expected[:4], expected[-4:], buf[:4], buf[-4:], i))
gt.wait() gt.wait()
def test_seek_on_buffered_pipe(self): def test_seek_on_buffered_pipe(self):
f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024) f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024)
self.assertEquals(f.tell(), 0) self.assertEqual(f.tell(), 0)
f.seek(0, 2) f.seek(0, 2)
self.assertEquals(f.tell(), 0) self.assertEqual(f.tell(), 0)
f.write('1234567890') f.write('1234567890')
f.seek(0, 2) f.seek(0, 2)
self.assertEquals(f.tell(), 10) self.assertEqual(f.tell(), 10)
f.seek(0) f.seek(0)
value = f.read(1) value = f.read(1)
self.assertEqual(value, '1') self.assertEqual(value, '1')
self.assertEquals(f.tell(), 1) self.assertEqual(f.tell(), 1)
value = f.read(1) value = f.read(1)
self.assertEqual(value, '2') self.assertEqual(value, '2')
self.assertEquals(f.tell(), 2) self.assertEqual(f.tell(), 2)
f.seek(0, 1) f.seek(0, 1)
self.assertEqual(f.readline(), '34567890') self.assertEqual(f.readline(), '34567890')
f.seek(-5, 1) f.seek(-5, 1)
@@ -711,7 +721,7 @@ class TestGreenPipe(LimitedTestCase):
f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024) f = greenio.GreenPipe(self.tempdir + "/TestFile", 'w+', 1024)
f.write('1234567890') f.write('1234567890')
f.truncate(9) f.truncate(9)
self.assertEquals(f.tell(), 9) self.assertEqual(f.tell(), 9)
class TestGreenIoLong(LimitedTestCase): class TestGreenIoLong(LimitedTestCase):
@@ -762,7 +772,7 @@ class TestGreenIoLong(LimitedTestCase):
bufsized(client, size=sendsize) bufsized(client, size=sendsize)
else: else:
bufsized(client) bufsized(client)
client.sendall(s2b('*') * sendsize) client.sendall(b'*' * sendsize)
client.close() client.close()
server_coro.wait() server_coro.wait()
listener.close() listener.close()
@@ -831,7 +841,7 @@ class TestGreenIoStarvation(LimitedTestCase):
bufsized(client, size=sendsize) bufsized(client, size=sendsize)
for i in range(sendloops): for i in range(sendloops):
client.sendall(s2b('*') * sendsize) client.sendall(b'*' * sendsize)
client.close() client.close()
os._exit(0) os._exit(0)
@@ -864,12 +874,15 @@ class TestGreenIoStarvation(LimitedTestCase):
# assert that the last task started before the first task ended # assert that the last task started before the first task ended
# (our no-starvation condition) # (our no-starvation condition)
assert starttimes[-1] < endtimes[0], "Not overlapping: starts %s ends %s" % (starttimes, endtimes) assert starttimes[-1] < endtimes[0], \
"Not overlapping: starts %s ends %s" % (starttimes, endtimes)
maxstartdiff = starttimes[-1] - starttimes[0] maxstartdiff = starttimes[-1] - starttimes[0]
assert maxstartdiff * 2 < runlengths[0], "Largest difference in starting times more than twice the shortest running time!" assert maxstartdiff * 2 < runlengths[0], \
assert runlengths[0] * 2 > runlengths[-1], "Longest runtime more than twice as long as shortest!" "Largest difference in starting times more than twice the shortest running time!"
assert runlengths[0] * 2 > runlengths[-1], \
"Longest runtime more than twice as long as shortest!"
def test_set_nonblocking(): def test_set_nonblocking():

108
tests/greenpool_test.py
View File

@@ -1,11 +1,9 @@
import gc import gc
import itertools
import os import os
import random import random
import eventlet import eventlet
from eventlet import debug from eventlet import hubs, greenpool, event, pools
from eventlet import hubs, greenpool, coros, event
from eventlet.support import greenlets as greenlet, six from eventlet.support import greenlets as greenlet, six
import tests import tests
@@ -17,7 +15,7 @@ def passthru(a):
def passthru2(a, b): def passthru2(a, b):
eventlet.sleep(0.01) eventlet.sleep(0.01)
return a,b return a, b
def raiser(exc): def raiser(exc):
@@ -31,24 +29,28 @@ class GreenPool(tests.LimitedTestCase):
for i in range(10): for i in range(10):
waiters.append(p.spawn(passthru, i)) waiters.append(p.spawn(passthru, i))
results = [waiter.wait() for waiter in waiters] results = [waiter.wait() for waiter in waiters]
self.assertEquals(results, list(range(10))) self.assertEqual(results, list(range(10)))
def test_spawn_n(self): def test_spawn_n(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
results_closure = [] results_closure = []
def do_something(a): def do_something(a):
eventlet.sleep(0.01) eventlet.sleep(0.01)
results_closure.append(a) results_closure.append(a)
for i in range(10): for i in range(10):
p.spawn(do_something, i) p.spawn(do_something, i)
p.waitall() p.waitall()
self.assertEquals(results_closure, list(range(10))) self.assertEqual(results_closure, list(range(10)))
def test_waiting(self): def test_waiting(self):
pool = greenpool.GreenPool(1) pool = greenpool.GreenPool(1)
done = event.Event() done = event.Event()
def consume(): def consume():
done.wait() done.wait()
def waiter(pool): def waiter(pool):
gt = pool.spawn(consume) gt = pool.spawn(consume)
gt.wait() gt.wait()
@@ -74,9 +76,11 @@ class GreenPool(tests.LimitedTestCase):
def test_multiple_coros(self): def test_multiple_coros(self):
evt = event.Event() evt = event.Event()
results = [] results = []
def producer(): def producer():
results.append('prod') results.append('prod')
evt.send() evt.send()
def consumer(): def consumer():
results.append('cons1') results.append('cons1')
evt.wait() evt.wait()
@@ -86,25 +90,29 @@ class GreenPool(tests.LimitedTestCase):
done = pool.spawn(consumer) done = pool.spawn(consumer)
pool.spawn_n(producer) pool.spawn_n(producer)
done.wait() done.wait()
self.assertEquals(['cons1', 'prod', 'cons2'], results) self.assertEqual(['cons1', 'prod', 'cons2'], results)
def test_timer_cancel(self): def test_timer_cancel(self):
# this test verifies that local timers are not fired # this test verifies that local timers are not fired
# outside of the context of the spawn # outside of the context of the spawn
timer_fired = [] timer_fired = []
def fire_timer(): def fire_timer():
timer_fired.append(True) timer_fired.append(True)
def some_work(): def some_work():
hubs.get_hub().schedule_call_local(0, fire_timer) hubs.get_hub().schedule_call_local(0, fire_timer)
pool = greenpool.GreenPool(2) pool = greenpool.GreenPool(2)
worker = pool.spawn(some_work) worker = pool.spawn(some_work)
worker.wait() worker.wait()
eventlet.sleep(0) eventlet.sleep(0)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(timer_fired, []) self.assertEqual(timer_fired, [])
def test_reentrant(self): def test_reentrant(self):
pool = greenpool.GreenPool(1) pool = greenpool.GreenPool(1)
def reenter(): def reenter():
waiter = pool.spawn(lambda a: a, 'reenter') waiter = pool.spawn(lambda a: a, 'reenter')
self.assertEqual('reenter', waiter.wait()) self.assertEqual('reenter', waiter.wait())
@@ -113,17 +121,20 @@ class GreenPool(tests.LimitedTestCase):
outer_waiter.wait() outer_waiter.wait()
evt = event.Event() evt = event.Event()
def reenter_async(): def reenter_async():
pool.spawn_n(lambda a: a, 'reenter') pool.spawn_n(lambda a: a, 'reenter')
evt.send('done') evt.send('done')
pool.spawn_n(reenter_async) pool.spawn_n(reenter_async)
self.assertEquals('done', evt.wait()) self.assertEqual('done', evt.wait())
def assert_pool_has_free(self, pool, num_free): def assert_pool_has_free(self, pool, num_free):
self.assertEquals(pool.free(), num_free) self.assertEqual(pool.free(), num_free)
def wait_long_time(e): def wait_long_time(e):
e.wait() e.wait()
timer = eventlet.Timeout(1) timer = eventlet.Timeout(1)
try: try:
evt = event.Event() evt = event.Event()
@@ -150,12 +161,14 @@ class GreenPool(tests.LimitedTestCase):
def test_resize(self): def test_resize(self):
pool = greenpool.GreenPool(2) pool = greenpool.GreenPool(2)
evt = event.Event() evt = event.Event()
def wait_long_time(e): def wait_long_time(e):
e.wait() e.wait()
pool.spawn(wait_long_time, evt) pool.spawn(wait_long_time, evt)
pool.spawn(wait_long_time, evt) pool.spawn(wait_long_time, evt)
self.assertEquals(pool.free(), 0) self.assertEqual(pool.free(), 0)
self.assertEquals(pool.running(), 2) self.assertEqual(pool.running(), 2)
self.assert_pool_has_free(pool, 0) self.assert_pool_has_free(pool, 0)
# verify that the pool discards excess items put into it # verify that the pool discards excess items put into it
@@ -167,28 +180,28 @@ class GreenPool(tests.LimitedTestCase):
eventlet.sleep(0) eventlet.sleep(0)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(pool.free(), 1) self.assertEqual(pool.free(), 1)
self.assertEquals(pool.running(), 0) self.assertEqual(pool.running(), 0)
self.assert_pool_has_free(pool, 1) self.assert_pool_has_free(pool, 1)
# resize larger and assert that there are more free items # resize larger and assert that there are more free items
pool.resize(2) pool.resize(2)
self.assertEquals(pool.free(), 2) self.assertEqual(pool.free(), 2)
self.assertEquals(pool.running(), 0) self.assertEqual(pool.running(), 0)
self.assert_pool_has_free(pool, 2) self.assert_pool_has_free(pool, 2)
def test_pool_smash(self): def test_pool_smash(self):
# The premise is that a coroutine in a Pool tries to get a token out # The premise is that a coroutine in a Pool tries to get a token out
# of a token pool but times out before getting the token. We verify # of a token pool but times out before getting the token. We verify
# that neither pool is adversely affected by this situation. # that neither pool is adversely affected by this situation.
from eventlet import pools
pool = greenpool.GreenPool(1) pool = greenpool.GreenPool(1)
tp = pools.TokenPool(max_size=1) tp = pools.TokenPool(max_size=1)
token = tp.get() # empty out the pool tp.get() # empty out the pool
def do_receive(tp): def do_receive(tp):
timer = eventlet.Timeout(0, RuntimeError()) timer = eventlet.Timeout(0, RuntimeError())
try: try:
t = tp.get() tp.get()
self.fail("Shouldn't have recieved anything from the pool") self.fail("Shouldn't have recieved anything from the pool")
except RuntimeError: except RuntimeError:
return 'timed out' return 'timed out'
@@ -198,7 +211,7 @@ class GreenPool(tests.LimitedTestCase):
# the spawn makes the token pool expect that coroutine, but then # the spawn makes the token pool expect that coroutine, but then
# immediately cuts bait # immediately cuts bait
e1 = pool.spawn(do_receive, tp) e1 = pool.spawn(do_receive, tp)
self.assertEquals(e1.wait(), 'timed out') self.assertEqual(e1.wait(), 'timed out')
# the pool can get some random item back # the pool can get some random item back
def send_wakeup(tp): def send_wakeup(tp):
@@ -210,18 +223,20 @@ class GreenPool(tests.LimitedTestCase):
def resume(): def resume():
return 'resumed' return 'resumed'
e2 = pool.spawn(resume) e2 = pool.spawn(resume)
self.assertEquals(e2.wait(), 'resumed') self.assertEqual(e2.wait(), 'resumed')
# we should be able to get out the thing we put in there, too # we should be able to get out the thing we put in there, too
self.assertEquals(tp.get(), 'wakeup') self.assertEqual(tp.get(), 'wakeup')
gt.wait() gt.wait()
def test_spawn_n_2(self): def test_spawn_n_2(self):
p = greenpool.GreenPool(2) p = greenpool.GreenPool(2)
self.assertEqual(p.free(), 2) self.assertEqual(p.free(), 2)
r = [] r = []
def foo(a): def foo(a):
r.append(a) r.append(a)
gt = p.spawn(foo, 1) gt = p.spawn(foo, 1)
self.assertEqual(p.free(), 1) self.assertEqual(p.free(), 1)
gt.wait() gt.wait()
@@ -239,9 +254,9 @@ class GreenPool(tests.LimitedTestCase):
self.assertEqual(r, [1]) self.assertEqual(r, [1])
p.spawn_n(foo, 4) p.spawn_n(foo, 4)
self.assertEqual(set(r), set([1,2,3])) self.assertEqual(set(r), set([1, 2, 3]))
eventlet.sleep(0) eventlet.sleep(0)
self.assertEqual(set(r), set([1,2,3,4])) self.assertEqual(set(r), set([1, 2, 3, 4]))
def test_exceptions(self): def test_exceptions(self):
p = greenpool.GreenPool(2) p = greenpool.GreenPool(2)
@@ -259,7 +274,7 @@ class GreenPool(tests.LimitedTestCase):
def test_imap(self): def test_imap(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
result_list = list(p.imap(passthru, range(10))) result_list = list(p.imap(passthru, range(10)))
self.assertEquals(result_list, list(range(10))) self.assertEqual(result_list, list(range(10)))
def test_empty_imap(self): def test_empty_imap(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
@@ -269,38 +284,40 @@ class GreenPool(tests.LimitedTestCase):
def test_imap_nonefunc(self): def test_imap_nonefunc(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
result_list = list(p.imap(None, range(10))) result_list = list(p.imap(None, range(10)))
self.assertEquals(result_list, [(x,) for x in range(10)]) self.assertEqual(result_list, [(x,) for x in range(10)])
def test_imap_multi_args(self): def test_imap_multi_args(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
result_list = list(p.imap(passthru2, range(10), range(10, 20))) result_list = list(p.imap(passthru2, range(10), range(10, 20)))
self.assertEquals(result_list, list(zip(range(10), range(10,20)))) self.assertEqual(result_list, list(zip(range(10), range(10, 20))))
def test_imap_raises(self): def test_imap_raises(self):
# testing the case where the function raises an exception; # testing the case where the function raises an exception;
# both that the caller sees that exception, and that the iterator # both that the caller sees that exception, and that the iterator
# continues to be usable to get the rest of the items # continues to be usable to get the rest of the items
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
def raiser(item): def raiser(item):
if item == 1 or item == 7: if item == 1 or item == 7:
raise RuntimeError("intentional error") raise RuntimeError("intentional error")
else: else:
return item return item
it = p.imap(raiser, range(10)) it = p.imap(raiser, range(10))
results = [] results = []
while True: while True:
try: try:
results.append(it.next()) results.append(six.next(it))
except RuntimeError: except RuntimeError:
results.append('r') results.append('r')
except StopIteration: except StopIteration:
break break
self.assertEquals(results, [0,'r',2,3,4,5,6,'r',8,9]) self.assertEqual(results, [0, 'r', 2, 3, 4, 5, 6, 'r', 8, 9])
def test_starmap(self): def test_starmap(self):
p = greenpool.GreenPool(4) p = greenpool.GreenPool(4)
result_list = list(p.starmap(passthru, [(x,) for x in range(10)])) result_list = list(p.starmap(passthru, [(x,) for x in range(10)]))
self.assertEquals(result_list, list(range(10))) self.assertEqual(result_list, list(range(10)))
def test_waitall_on_nothing(self): def test_waitall_on_nothing(self):
p = greenpool.GreenPool() p = greenpool.GreenPool()
@@ -318,7 +335,7 @@ class GreenPile(tests.LimitedTestCase):
for i in range(10): for i in range(10):
p.spawn(passthru, i) p.spawn(passthru, i)
result_list = list(p) result_list = list(p)
self.assertEquals(result_list, list(range(10))) self.assertEqual(result_list, list(range(10)))
def test_pile_spawn_times_out(self): def test_pile_spawn_times_out(self):
p = greenpool.GreenPile(4) p = greenpool.GreenPile(4)
@@ -329,28 +346,32 @@ class GreenPile(tests.LimitedTestCase):
self.assertRaises(eventlet.Timeout, p.spawn, passthru, "time out") self.assertRaises(eventlet.Timeout, p.spawn, passthru, "time out")
# verify that the spawn breakage didn't interrupt the sequence # verify that the spawn breakage didn't interrupt the sequence
# and terminates properly # and terminates properly
for i in range(4,10): for i in range(4, 10):
p.spawn(passthru, i) p.spawn(passthru, i)
self.assertEquals(list(p), list(range(10))) self.assertEqual(list(p), list(range(10)))
def test_constructing_from_pool(self): def test_constructing_from_pool(self):
pool = greenpool.GreenPool(2) pool = greenpool.GreenPool(2)
pile1 = greenpool.GreenPile(pool) pile1 = greenpool.GreenPile(pool)
pile2 = greenpool.GreenPile(pool) pile2 = greenpool.GreenPile(pool)
def bunch_of_work(pile, unique): def bunch_of_work(pile, unique):
for i in range(10): for i in range(10):
pile.spawn(passthru, i + unique) pile.spawn(passthru, i + unique)
eventlet.spawn(bunch_of_work, pile1, 0) eventlet.spawn(bunch_of_work, pile1, 0)
eventlet.spawn(bunch_of_work, pile2, 100) eventlet.spawn(bunch_of_work, pile2, 100)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(list(pile2), list(range(100,110))) self.assertEqual(list(pile2), list(range(100, 110)))
self.assertEquals(list(pile1), list(range(10))) self.assertEqual(list(pile1), list(range(10)))
class StressException(Exception): class StressException(Exception):
pass pass
r = random.Random(0) r = random.Random(0)
def pressure(arg): def pressure(arg):
while r.random() < 0.5: while r.random() < 0.5:
eventlet.sleep(r.random() * 0.001) eventlet.sleep(r.random() * 0.001)
@@ -359,18 +380,22 @@ def pressure(arg):
else: else:
raise StressException(arg) raise StressException(arg)
def passthru(arg): def passthru(arg):
while r.random() < 0.5: while r.random() < 0.5:
eventlet.sleep(r.random() * 0.001) eventlet.sleep(r.random() * 0.001)
return arg return arg
class Stress(tests.LimitedTestCase): class Stress(tests.LimitedTestCase):
# tests will take extra-long # tests will take extra-long
TEST_TIMEOUT=60 TEST_TIMEOUT = 60
@tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES') @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
def spawn_order_check(self, concurrency): def spawn_order_check(self, concurrency):
# checks that piles are strictly ordered # checks that piles are strictly ordered
p = greenpool.GreenPile(concurrency) p = greenpool.GreenPile(concurrency)
def makework(count, unique): def makework(count, unique):
for i in six.moves.range(count): for i in six.moves.range(count):
token = (unique, i) token = (unique, i)
@@ -380,13 +405,13 @@ class Stress(tests.LimitedTestCase):
eventlet.spawn(makework, iters, 1) eventlet.spawn(makework, iters, 1)
eventlet.spawn(makework, iters, 2) eventlet.spawn(makework, iters, 2)
eventlet.spawn(makework, iters, 3) eventlet.spawn(makework, iters, 3)
p.spawn(pressure, (0,0)) p.spawn(pressure, (0, 0))
latest = [-1] * 4 latest = [-1] * 4
received = 0 received = 0
it = iter(p) it = iter(p)
while True: while True:
try: try:
i = it.next() i = six.next(it)
except StressException as exc: except StressException as exc:
i = exc.args[0] i = exc.args[0]
except StopIteration: except StopIteration:
@@ -398,7 +423,7 @@ class Stress(tests.LimitedTestCase):
self.assert_(latest[unique] < order) self.assert_(latest[unique] < order)
latest[unique] = order latest[unique] = order
for l in latest[1:]: for l in latest[1:]:
self.assertEquals(l, iters - 1) self.assertEqual(l, iters - 1)
@tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES') @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
def test_ordering_5(self): def test_ordering_5(self):
@@ -417,7 +442,7 @@ class Stress(tests.LimitedTestCase):
latest = -1 latest = -1
while True: while True:
try: try:
i = it.next() i = six.next(it)
except StopIteration: except StopIteration:
break break
@@ -433,7 +458,7 @@ class Stress(tests.LimitedTestCase):
objs_created = len(gc.get_objects()) - initial_obj_count objs_created = len(gc.get_objects()) - initial_obj_count
self.assert_(objs_created < 25 * concurrency, objs_created) self.assert_(objs_created < 25 * concurrency, objs_created)
# make sure we got to the end # make sure we got to the end
self.assertEquals(latest, count - 1) self.assertEqual(latest, count - 1)
@tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES') @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
def test_imap_50(self): def test_imap_50(self):
@@ -445,7 +470,6 @@ class Stress(tests.LimitedTestCase):
@tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES') @tests.skip_unless(os.environ.get('RUN_STRESS_TESTS') == 'YES')
def test_with_intpool(self): def test_with_intpool(self):
from eventlet import pools
class IntPool(pools.Pool): class IntPool(pools.Pool):
def create(self): def create(self):
self.current_integer = getattr(self, 'current_integer', 0) + 1 self.current_integer = getattr(self, 'current_integer', 0) + 1

26
tests/greenthread_test.py
View File

@@ -27,22 +27,22 @@ class Spawn(LimitedTestCase, Asserts):
def test_simple(self): def test_simple(self):
gt = greenthread.spawn(passthru, 1, b=2) gt = greenthread.spawn(passthru, 1, b=2)
self.assertEquals(gt.wait(), ((1,),{'b':2})) self.assertEqual(gt.wait(), ((1,),{'b':2}))
self.assertEquals(_g_results, [((1,),{'b':2})]) self.assertEqual(_g_results, [((1,),{'b':2})])
def test_n(self): def test_n(self):
gt = greenthread.spawn_n(passthru, 2, b=3) gt = greenthread.spawn_n(passthru, 2, b=3)
self.assert_(not gt.dead) self.assert_(not gt.dead)
greenthread.sleep(0) greenthread.sleep(0)
self.assert_(gt.dead) self.assert_(gt.dead)
self.assertEquals(_g_results, [((2,),{'b':3})]) self.assertEqual(_g_results, [((2,),{'b':3})])
def test_kill(self): def test_kill(self):
gt = greenthread.spawn(passthru, 6) gt = greenthread.spawn(passthru, 6)
greenthread.kill(gt) greenthread.kill(gt)
self.assert_dead(gt) self.assert_dead(gt)
greenthread.sleep(0.001) greenthread.sleep(0.001)
self.assertEquals(_g_results, []) self.assertEqual(_g_results, [])
greenthread.kill(gt) greenthread.kill(gt)
self.assert_dead(gt) self.assert_dead(gt)
@@ -51,7 +51,7 @@ class Spawn(LimitedTestCase, Asserts):
gt.kill() gt.kill()
self.assert_dead(gt) self.assert_dead(gt)
greenthread.sleep(0.001) greenthread.sleep(0.001)
self.assertEquals(_g_results, []) self.assertEqual(_g_results, [])
gt.kill() gt.kill()
self.assert_dead(gt) self.assert_dead(gt)
@@ -60,7 +60,7 @@ class Spawn(LimitedTestCase, Asserts):
greenthread.kill(gt) greenthread.kill(gt)
self.assert_dead(gt) self.assert_dead(gt)
greenthread.sleep(0.001) greenthread.sleep(0.001)
self.assertEquals(_g_results, []) self.assertEqual(_g_results, [])
greenthread.kill(gt) greenthread.kill(gt)
self.assert_dead(gt) self.assert_dead(gt)
@@ -72,8 +72,8 @@ class Spawn(LimitedTestCase, Asserts):
results.append(kw) results.append(kw)
gt = greenthread.spawn(passthru, 5) gt = greenthread.spawn(passthru, 5)
gt.link(link_func, 4, b=5) gt.link(link_func, 4, b=5)
self.assertEquals(gt.wait(), ((5,), {})) self.assertEqual(gt.wait(), ((5,), {}))
self.assertEquals(results, [gt, (4,), {'b':5}]) self.assertEqual(results, [gt, (4,), {'b':5}])
def test_link_after_exited(self): def test_link_after_exited(self):
results = [] results = []
@@ -82,9 +82,9 @@ class Spawn(LimitedTestCase, Asserts):
results.append(a) results.append(a)
results.append(kw) results.append(kw)
gt = greenthread.spawn(passthru, 5) gt = greenthread.spawn(passthru, 5)
self.assertEquals(gt.wait(), ((5,), {})) self.assertEqual(gt.wait(), ((5,), {}))
gt.link(link_func, 4, b=5) gt.link(link_func, 4, b=5)
self.assertEquals(results, [gt, (4,), {'b':5}]) self.assertEqual(results, [gt, (4,), {'b':5}])
def test_link_relinks(self): def test_link_relinks(self):
# test that linking in a linked func doesn't cause infinite recursion. # test that linking in a linked func doesn't cause infinite recursion.
@@ -99,12 +99,12 @@ class Spawn(LimitedTestCase, Asserts):
gt = greenthread.spawn(passthru) gt = greenthread.spawn(passthru)
gt.link(link_func) gt.link(link_func)
gt.wait() gt.wait()
self.assertEquals(called, [True]) self.assertEqual(called, [True])
class SpawnAfter(Spawn): class SpawnAfter(Spawn):
def test_basic(self): def test_basic(self):
gt = greenthread.spawn_after(0.1, passthru, 20) gt = greenthread.spawn_after(0.1, passthru, 20)
self.assertEquals(gt.wait(), ((20,), {})) self.assertEqual(gt.wait(), ((20,), {}))
def test_cancel(self): def test_cancel(self):
gt = greenthread.spawn_after(0.1, passthru, 21) gt = greenthread.spawn_after(0.1, passthru, 21)
@@ -115,7 +115,7 @@ class SpawnAfter(Spawn):
gt = greenthread.spawn_after(0, waiter, 22) gt = greenthread.spawn_after(0, waiter, 22)
greenthread.sleep(0) greenthread.sleep(0)
gt.cancel() gt.cancel()
self.assertEquals(gt.wait(), 22) self.assertEqual(gt.wait(), 22)
def test_kill_already_started(self): def test_kill_already_started(self):
gt = greenthread.spawn_after(0, waiter, 22) gt = greenthread.spawn_after(0, waiter, 22)

44
tests/hub_test.py
View File

@@ -1,12 +1,12 @@
from __future__ import with_statement from __future__ import with_statement
import sys import sys
import tests
from tests import LimitedTestCase, main, skip_with_pyevent, skip_if_no_itimer, skip_unless from tests import LimitedTestCase, main, skip_with_pyevent, skip_if_no_itimer, skip_unless
from tests.patcher_test import ProcessBase from tests.patcher_test import ProcessBase
import time import time
import eventlet import eventlet
from eventlet import hubs from eventlet import hubs
from eventlet.green import socket
from eventlet.event import Event from eventlet.event import Event
from eventlet.semaphore import Semaphore from eventlet.semaphore import Semaphore
from eventlet.support import greenlets, six from eventlet.support import greenlets, six
@@ -108,7 +108,7 @@ class TestScheduleCall(LimitedTestCase):
hubs.get_hub().schedule_call_global(DELAY, lst.append, 2) hubs.get_hub().schedule_call_global(DELAY, lst.append, 2)
while len(lst) < 3: while len(lst) < 3:
eventlet.sleep(DELAY) eventlet.sleep(DELAY)
self.assertEquals(lst, [1, 2, 3]) self.assertEqual(lst, [1, 2, 3])
class TestDebug(LimitedTestCase): class TestDebug(LimitedTestCase):
@@ -250,6 +250,8 @@ class TestHubBlockingDetector(LimitedTestCase):
class TestSuspend(LimitedTestCase): class TestSuspend(LimitedTestCase):
TEST_TIMEOUT = 3 TEST_TIMEOUT = 3
longMessage = True
maxDiff = None
def test_suspend_doesnt_crash(self): def test_suspend_doesnt_crash(self):
import os import os
@@ -279,8 +281,8 @@ except eventlet.Timeout:
os.kill(p.pid, signal.SIGSTOP) # suspend and resume to generate EINTR os.kill(p.pid, signal.SIGSTOP) # suspend and resume to generate EINTR
os.kill(p.pid, signal.SIGCONT) os.kill(p.pid, signal.SIGCONT)
output, _ = p.communicate() output, _ = p.communicate()
lines = [l for l in output.split("\n") if l] lines = output.decode('utf-8', 'replace').splitlines()
self.assert_("exited correctly" in lines[-1]) self.assert_("exited correctly" in lines[-1], output)
shutil.rmtree(self.tempdir) shutil.rmtree(self.tempdir)
@@ -293,39 +295,13 @@ class TestBadFilenos(LimitedTestCase):
self.assertRaises(ValueError, select.select, [-1], [], []) self.assertRaises(ValueError, select.select, [-1], [], [])
class TestFork(ProcessBase): class TestFork(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_fork(self): def test_fork(self):
new_mod = """ output = tests.run_python('tests/hub_test_fork.py')
import os lines = output.splitlines()
import eventlet self.assertEqual(lines, ["accept blocked", "child died ok"], output)
server = eventlet.listen(('localhost', 12345))
t = eventlet.Timeout(0.01)
try:
new_sock, address = server.accept()
except eventlet.Timeout as t:
pass
pid = os.fork()
if not pid:
t = eventlet.Timeout(0.1)
try:
new_sock, address = server.accept()
except eventlet.Timeout as t:
print("accept blocked")
else:
kpid, status = os.wait()
assert kpid == pid
assert status == 0
print("child died ok")
"""
self.write_to_tempfile("newmod", new_mod)
output, lines = self.launch_subprocess('newmod.py')
self.assertEqual(len(lines), 3, output)
self.assert_("accept blocked" in lines[0])
self.assert_("child died ok" in lines[1])
class TestDeadRunLoop(LimitedTestCase): class TestDeadRunLoop(LimitedTestCase):

25
tests/hub_test_fork.py Normal file
View File

@@ -0,0 +1,25 @@
# no standard tests in this file, ignore
__test__ = False
if __name__ == '__main__':
import os
import eventlet
server = eventlet.listen(('localhost', 12345))
t = eventlet.Timeout(0.01)
try:
new_sock, address = server.accept()
except eventlet.Timeout as t:
pass
pid = os.fork()
if not pid:
t = eventlet.Timeout(0.1)
try:
new_sock, address = server.accept()
except eventlet.Timeout as t:
print("accept blocked")
else:
kpid, status = os.wait()
assert kpid == pid
assert status == 0
print("child died ok")

8
tests/mysqldb_test.py
View File

@@ -122,14 +122,18 @@ class TestMySQLdb(LimitedTestCase):
gt = eventlet.spawn(tick) gt = eventlet.spawn(tick)
curs.execute("select 1") curs.execute("select 1")
rows = curs.fetchall() rows = curs.fetchall()
self.assertEqual(rows, ((1L,),)) self.assertEqual(len(rows), 1)
self.assertEqual(len(rows[0]), 1)
self.assertEqual(rows[0][0], 1)
self.assert_(counter[0] > 0, counter[0]) self.assert_(counter[0] > 0, counter[0])
gt.kill() gt.kill()
def assert_cursor_works(self, cursor): def assert_cursor_works(self, cursor):
cursor.execute("select 1") cursor.execute("select 1")
rows = cursor.fetchall() rows = cursor.fetchall()
self.assertEqual(rows, ((1L,),)) self.assertEqual(len(rows), 1)
self.assertEqual(len(rows[0]), 1)
self.assertEqual(rows[0][0], 1)
self.assert_cursor_yields(cursor) self.assert_cursor_yields(cursor)
def assert_connection_works(self, conn): def assert_connection_works(self, conn):

50
tests/pools_test.py
View File

@@ -28,17 +28,17 @@ class TestIntPool(TestCase):
# with self.pool.some_api_name() as thing: # with self.pool.some_api_name() as thing:
# # do stuff # # do stuff
self.assertEquals(self.pool.get(), 1) self.assertEqual(self.pool.get(), 1)
self.assertEquals(self.pool.get(), 2) self.assertEqual(self.pool.get(), 2)
self.assertEquals(self.pool.get(), 3) self.assertEqual(self.pool.get(), 3)
self.assertEquals(self.pool.get(), 4) self.assertEqual(self.pool.get(), 4)
def test_free(self): def test_free(self):
self.assertEquals(self.pool.free(), 4) self.assertEqual(self.pool.free(), 4)
gotten = self.pool.get() gotten = self.pool.get()
self.assertEquals(self.pool.free(), 3) self.assertEqual(self.pool.free(), 3)
self.pool.put(gotten) self.pool.put(gotten)
self.assertEquals(self.pool.free(), 4) self.assertEqual(self.pool.free(), 4)
def test_exhaustion(self): def test_exhaustion(self):
waiter = Queue(0) waiter = Queue(0)
@@ -53,7 +53,7 @@ class TestIntPool(TestCase):
one, two, three, four = ( one, two, three, four = (
self.pool.get(), self.pool.get(), self.pool.get(), self.pool.get()) self.pool.get(), self.pool.get(), self.pool.get(), self.pool.get())
self.assertEquals(self.pool.free(), 0) self.assertEqual(self.pool.free(), 0)
# Let consumer run; nothing will be in the pool, so he will wait # Let consumer run; nothing will be in the pool, so he will wait
eventlet.sleep(0) eventlet.sleep(0)
@@ -62,7 +62,7 @@ class TestIntPool(TestCase):
self.pool.put(one) self.pool.put(one)
# wait for the consumer # wait for the consumer
self.assertEquals(waiter.get(), one) self.assertEqual(waiter.get(), one)
def test_blocks_on_pool(self): def test_blocks_on_pool(self):
waiter = Queue(0) waiter = Queue(0)
@@ -72,7 +72,7 @@ class TestIntPool(TestCase):
self.pool.get() self.pool.get()
self.pool.get() self.pool.get()
# No one should be waiting yet. # No one should be waiting yet.
self.assertEquals(self.pool.waiting(), 0) self.assertEqual(self.pool.waiting(), 0)
# The call to the next get will unschedule this routine. # The call to the next get will unschedule this routine.
self.pool.get() self.pool.get()
# So this put should never be called. # So this put should never be called.
@@ -81,13 +81,13 @@ class TestIntPool(TestCase):
killable = eventlet.spawn(greedy) killable = eventlet.spawn(greedy)
# no one should be waiting yet. # no one should be waiting yet.
self.assertEquals(self.pool.waiting(), 0) self.assertEqual(self.pool.waiting(), 0)
## Wait for greedy ## Wait for greedy
eventlet.sleep(0) eventlet.sleep(0)
## Greedy should be blocking on the last get ## Greedy should be blocking on the last get
self.assertEquals(self.pool.waiting(), 1) self.assertEqual(self.pool.waiting(), 1)
## Send will never be called, so balance should be 0. ## Send will never be called, so balance should be 0.
self.assertFalse(not waiter.full()) self.assertFalse(not waiter.full())
@@ -100,8 +100,8 @@ class TestIntPool(TestCase):
one, two = self.pool.get(), self.pool.get() one, two = self.pool.get(), self.pool.get()
self.pool.put(one) self.pool.put(one)
self.pool.put(two) self.pool.put(two)
self.assertEquals(self.pool.get(), one) self.assertEqual(self.pool.get(), one)
self.assertEquals(self.pool.get(), two) self.assertEqual(self.pool.get(), two)
def test_putting_to_queue(self): def test_putting_to_queue(self):
timer = eventlet.Timeout(0.1) timer = eventlet.Timeout(0.1)
@@ -128,17 +128,17 @@ class TestIntPool(TestCase):
pool = IntPool(max_size=2) pool = IntPool(max_size=2)
a = pool.get() a = pool.get()
b = pool.get() b = pool.get()
self.assertEquals(pool.free(), 0) self.assertEqual(pool.free(), 0)
# verify that the pool discards excess items put into it # verify that the pool discards excess items put into it
pool.resize(1) pool.resize(1)
pool.put(a) pool.put(a)
pool.put(b) pool.put(b)
self.assertEquals(pool.free(), 1) self.assertEqual(pool.free(), 1)
# resize larger and assert that there are more free items # resize larger and assert that there are more free items
pool.resize(2) pool.resize(2)
self.assertEquals(pool.free(), 2) self.assertEqual(pool.free(), 2)
def test_create_contention(self): def test_create_contention(self):
creates = [0] creates = [0]
@@ -151,14 +151,14 @@ class TestIntPool(TestCase):
def do_get(): def do_get():
x = p.get() x = p.get()
self.assertEquals(x, "slept") self.assertEqual(x, "slept")
p.put(x) p.put(x)
gp = eventlet.GreenPool() gp = eventlet.GreenPool()
for i in six.moves.range(100): for i in six.moves.range(100):
gp.spawn_n(do_get) gp.spawn_n(do_get)
gp.waitall() gp.waitall()
self.assertEquals(creates[0], 4) self.assertEqual(creates[0], 4)
class TestAbstract(TestCase): class TestAbstract(TestCase):
@@ -176,11 +176,11 @@ class TestIntPool2(TestCase):
self.pool = IntPool(min_size=3, max_size=3) self.pool = IntPool(min_size=3, max_size=3)
def test_something(self): def test_something(self):
self.assertEquals(len(self.pool.free_items), 3) self.assertEqual(len(self.pool.free_items), 3)
## Cover the clause in get where we get from the free list instead of creating ## Cover the clause in get where we get from the free list instead of creating
## an item on get ## an item on get
gotten = self.pool.get() gotten = self.pool.get()
self.assertEquals(gotten, 1) self.assertEqual(gotten, 1)
class TestOrderAsStack(TestCase): class TestOrderAsStack(TestCase):
@@ -193,8 +193,8 @@ class TestOrderAsStack(TestCase):
one, two = self.pool.get(), self.pool.get() one, two = self.pool.get(), self.pool.get()
self.pool.put(one) self.pool.put(one)
self.pool.put(two) self.pool.put(two)
self.assertEquals(self.pool.get(), two) self.assertEqual(self.pool.get(), two)
self.assertEquals(self.pool.get(), one) self.assertEqual(self.pool.get(), one)
class RaisePool(pools.Pool): class RaisePool(pools.Pool):
@@ -208,9 +208,9 @@ class TestCreateRaises(TestCase):
self.pool = RaisePool(max_size=3) self.pool = RaisePool(max_size=3)
def test_it(self): def test_it(self):
self.assertEquals(self.pool.free(), 3) self.assertEqual(self.pool.free(), 3)
self.assertRaises(RuntimeError, self.pool.get) self.assertRaises(RuntimeError, self.pool.get)
self.assertEquals(self.pool.free(), 3) self.assertEqual(self.pool.free(), 3)
ALWAYS = RuntimeError('I always fail') ALWAYS = RuntimeError('I always fail')

8
tests/processes_test.py
View File

@@ -20,7 +20,7 @@ class TestEchoPool(LimitedTestCase):
result = proc.read() result = proc.read()
finally: finally:
self.pool.put(proc) self.pool.put(proc)
self.assertEquals(result, 'hello\n') self.assertEqual(result, 'hello\n')
@skip_on_windows @skip_on_windows
def test_read_eof(self): def test_read_eof(self):
@@ -34,7 +34,7 @@ class TestEchoPool(LimitedTestCase):
@skip_on_windows @skip_on_windows
def test_empty_echo(self): def test_empty_echo(self):
p = processes.Process('echo', ['-n']) p = processes.Process('echo', ['-n'])
self.assertEquals('', p.read()) self.assertEqual('', p.read())
self.assertRaises(processes.DeadProcess, p.read) self.assertRaises(processes.DeadProcess, p.read)
@@ -56,7 +56,7 @@ class TestCatPool(LimitedTestCase):
finally: finally:
self.pool.put(proc) self.pool.put(proc)
self.assertEquals(result, 'goodbye') self.assertEqual(result, 'goodbye')
@skip_on_windows @skip_on_windows
def test_write_to_dead(self): def test_write_to_dead(self):
@@ -95,7 +95,7 @@ class TestDyingProcessesLeavePool(LimitedTestCase):
try: try:
try: try:
result = proc.read() result = proc.read()
self.assertEquals(result, 'hello\n') self.assertEqual(result, 'hello\n')
result = proc.read() result = proc.read()
except processes.DeadProcess: except processes.DeadProcess:
pass pass

54
tests/queue_test.py
View File

@@ -14,12 +14,12 @@ class TestQueue(LimitedTestCase):
def test_send_first(self): def test_send_first(self):
q = eventlet.Queue() q = eventlet.Queue()
q.put('hi') q.put('hi')
self.assertEquals(q.get(), 'hi') self.assertEqual(q.get(), 'hi')
def test_send_last(self): def test_send_last(self):
q = eventlet.Queue() q = eventlet.Queue()
def waiter(q): def waiter(q):
self.assertEquals(q.get(), 'hi2') self.assertEqual(q.get(), 'hi2')
gt = eventlet.spawn(eventlet.with_timeout, 0.1, waiter, q) gt = eventlet.spawn(eventlet.with_timeout, 0.1, waiter, q)
eventlet.sleep(0) eventlet.sleep(0)
@@ -41,12 +41,12 @@ class TestQueue(LimitedTestCase):
gt = eventlet.spawn(putter, q) gt = eventlet.spawn(putter, q)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(results, ['a', 'b']) self.assertEqual(results, ['a', 'b'])
self.assertEquals(q.get(), 'a') self.assertEqual(q.get(), 'a')
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(results, ['a', 'b', 'c']) self.assertEqual(results, ['a', 'b', 'c'])
self.assertEquals(q.get(), 'b') self.assertEqual(q.get(), 'b')
self.assertEquals(q.get(), 'c') self.assertEqual(q.get(), 'c')
gt.wait() gt.wait()
def test_zero_max_size(self): def test_zero_max_size(self):
@@ -64,8 +64,8 @@ class TestQueue(LimitedTestCase):
eventlet.sleep(0) eventlet.sleep(0)
self.assert_(not evt.ready()) self.assert_(not evt.ready())
gt2 = eventlet.spawn(receiver, q) gt2 = eventlet.spawn(receiver, q)
self.assertEquals(gt2.wait(),'hi') self.assertEqual(gt2.wait(),'hi')
self.assertEquals(evt.wait(),'done') self.assertEqual(evt.wait(),'done')
gt.wait() gt.wait()
def test_resize_up(self): def test_resize_up(self):
@@ -90,10 +90,10 @@ class TestQueue(LimitedTestCase):
for i in range(5): for i in range(5):
q.put(i) q.put(i)
self.assertEquals(list(q.queue), list(range(5))) self.assertEqual(list(q.queue), list(range(5)))
q.resize(1) q.resize(1)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(list(q.queue), list(range(5))) self.assertEqual(list(q.queue), list(range(5)))
def test_resize_to_Unlimited(self): def test_resize_to_Unlimited(self):
q = eventlet.Queue(0) q = eventlet.Queue(0)
@@ -127,21 +127,21 @@ class TestQueue(LimitedTestCase):
results = set() results = set()
for i, gt in enumerate(gts): for i, gt in enumerate(gts):
results.add(gt.wait()) results.add(gt.wait())
self.assertEquals(results, set(sendings)) self.assertEqual(results, set(sendings))
def test_waiters_that_cancel(self): def test_waiters_that_cancel(self):
q = eventlet.Queue() q = eventlet.Queue()
gt = eventlet.spawn(do_bail, q) gt = eventlet.spawn(do_bail, q)
self.assertEquals(gt.wait(), 'timed out') self.assertEqual(gt.wait(), 'timed out')
q.put('hi') q.put('hi')
self.assertEquals(q.get(), 'hi') self.assertEqual(q.get(), 'hi')
def test_getting_before_sending(self): def test_getting_before_sending(self):
q = eventlet.Queue() q = eventlet.Queue()
gt = eventlet.spawn(q.put, 'sent') gt = eventlet.spawn(q.put, 'sent')
self.assertEquals(q.get(), 'sent') self.assertEqual(q.get(), 'sent')
gt.wait() gt.wait()
def test_two_waiters_one_dies(self): def test_two_waiters_one_dies(self):
@@ -153,8 +153,8 @@ class TestQueue(LimitedTestCase):
waiting = eventlet.spawn(waiter, q) waiting = eventlet.spawn(waiter, q)
eventlet.sleep(0) eventlet.sleep(0)
q.put('hi') q.put('hi')
self.assertEquals(dying.wait(), 'timed out') self.assertEqual(dying.wait(), 'timed out')
self.assertEquals(waiting.wait(), 'hi') self.assertEqual(waiting.wait(), 'hi')
def test_two_bogus_waiters(self): def test_two_bogus_waiters(self):
q = eventlet.Queue() q = eventlet.Queue()
@@ -162,20 +162,20 @@ class TestQueue(LimitedTestCase):
gt2 = eventlet.spawn(do_bail, q) gt2 = eventlet.spawn(do_bail, q)
eventlet.sleep(0) eventlet.sleep(0)
q.put('sent') q.put('sent')
self.assertEquals(gt1.wait(), 'timed out') self.assertEqual(gt1.wait(), 'timed out')
self.assertEquals(gt2.wait(), 'timed out') self.assertEqual(gt2.wait(), 'timed out')
self.assertEquals(q.get(), 'sent') self.assertEqual(q.get(), 'sent')
def test_waiting(self): def test_waiting(self):
q = eventlet.Queue() q = eventlet.Queue()
gt1 = eventlet.spawn(q.get) gt1 = eventlet.spawn(q.get)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(1, q.getting()) self.assertEqual(1, q.getting())
q.put('hi') q.put('hi')
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(0, q.getting()) self.assertEqual(0, q.getting())
self.assertEquals('hi', gt1.wait()) self.assertEqual('hi', gt1.wait())
self.assertEquals(0, q.getting()) self.assertEqual(0, q.getting())
def test_channel_send(self): def test_channel_send(self):
channel = eventlet.Queue(0) channel = eventlet.Queue(0)
@@ -222,7 +222,7 @@ class TestQueue(LimitedTestCase):
w2 = eventlet.spawn(c.get) w2 = eventlet.spawn(c.get)
w3 = eventlet.spawn(c.get) w3 = eventlet.spawn(c.get)
eventlet.sleep(0) eventlet.sleep(0)
self.assertEquals(c.getting(), 3) self.assertEqual(c.getting(), 3)
s1 = eventlet.spawn(c.put, 1) s1 = eventlet.spawn(c.put, 1)
s2 = eventlet.spawn(c.put, 2) s2 = eventlet.spawn(c.put, 2)
s3 = eventlet.spawn(c.put, 3) s3 = eventlet.spawn(c.put, 3)
@@ -230,10 +230,10 @@ class TestQueue(LimitedTestCase):
s1.wait() s1.wait()
s2.wait() s2.wait()
s3.wait() s3.wait()
self.assertEquals(c.getting(), 0) self.assertEqual(c.getting(), 0)
# NOTE: we don't guarantee that waiters are served in order # NOTE: we don't guarantee that waiters are served in order
results = sorted([w1.wait(), w2.wait(), w3.wait()]) results = sorted([w1.wait(), w2.wait(), w3.wait()])
self.assertEquals(results, [1,2,3]) self.assertEqual(results, [1,2,3])
def test_channel_sender_timing_out(self): def test_channel_sender_timing_out(self):
from eventlet import queue from eventlet import queue

14
tests/ssl_test.py
View File

@@ -46,7 +46,7 @@ class SSLTest(LimitedTestCase):
client = util.wrap_ssl(eventlet.connect(('127.0.0.1', sock.getsockname()[1]))) client = util.wrap_ssl(eventlet.connect(('127.0.0.1', sock.getsockname()[1])))
client.write(b'line 1\r\nline 2\r\n\r\n') client.write(b'line 1\r\nline 2\r\n\r\n')
self.assertEquals(client.read(8192), b'response') self.assertEqual(client.read(8192), b'response')
server_coro.wait() server_coro.wait()
@skip_if_no_ssl @skip_if_no_ssl
@@ -55,7 +55,7 @@ class SSLTest(LimitedTestCase):
sock, addr = listener.accept() sock, addr = listener.accept()
sock.read(8192) sock.read(8192)
try: try:
self.assertEquals(b"", sock.read(8192)) self.assertEqual(b"", sock.read(8192))
except greenio.SSL.ZeroReturnError: except greenio.SSL.ZeroReturnError:
pass pass
@@ -90,13 +90,13 @@ class SSLTest(LimitedTestCase):
def test_ssl_unwrap(self): def test_ssl_unwrap(self):
def serve(): def serve():
sock, addr = listener.accept() sock, addr = listener.accept()
self.assertEquals(sock.recv(6), b'before') self.assertEqual(sock.recv(6), b'before')
sock_ssl = util.wrap_ssl(sock, certificate_file, private_key_file, sock_ssl = util.wrap_ssl(sock, certificate_file, private_key_file,
server_side=True) server_side=True)
sock_ssl.do_handshake() sock_ssl.do_handshake()
self.assertEquals(sock_ssl.read(6), b'during') self.assertEqual(sock_ssl.read(6), b'during')
sock2 = sock_ssl.unwrap() sock2 = sock_ssl.unwrap()
self.assertEquals(sock2.recv(5), b'after') self.assertEqual(sock2.recv(5), b'after')
sock2.close() sock2.close()
listener = eventlet.listen(('127.0.0.1', 0)) listener = eventlet.listen(('127.0.0.1', 0))
@@ -160,8 +160,8 @@ class SSLTest(LimitedTestCase):
listener = listen_ssl_socket(('', 0)) listener = listen_ssl_socket(('', 0))
eventlet.spawn(serve, listener) eventlet.spawn(serve, listener)
client = ssl(eventlet.connect(('localhost', listener.getsockname()[1]))) client = ssl(eventlet.connect(('localhost', listener.getsockname()[1])))
self.assertEquals(client.read(1024), b'content') self.assertEqual(client.read(1024), b'content')
self.assertEquals(client.read(1024), b'') self.assertEqual(client.read(1024), b'')
@skip_if_no_ssl @skip_if_no_ssl
def test_regression_gh_17(self): def test_regression_gh_17(self):

50
tests/test__coros_queue.py
View File

@@ -11,7 +11,7 @@ class TestQueue(LimitedTestCase):
def test_send_first(self): def test_send_first(self):
q = coros.queue() q = coros.queue()
q.send('hi') q.send('hi')
self.assertEquals(q.wait(), 'hi') self.assertEqual(q.wait(), 'hi')
@silence_warnings @silence_warnings
def test_send_exception_first(self): def test_send_exception_first(self):
@@ -24,7 +24,7 @@ class TestQueue(LimitedTestCase):
q = coros.queue() q = coros.queue()
def waiter(q): def waiter(q):
timer = eventlet.Timeout(0.1) timer = eventlet.Timeout(0.1)
self.assertEquals(q.wait(), 'hi2') self.assertEqual(q.wait(), 'hi2')
timer.cancel() timer.cancel()
spawn(waiter, q) spawn(waiter, q)
@@ -47,12 +47,12 @@ class TestQueue(LimitedTestCase):
spawn(putter, q) spawn(putter, q)
sleep(0) sleep(0)
self.assertEquals(results, ['a', 'b']) self.assertEqual(results, ['a', 'b'])
self.assertEquals(q.wait(), 'a') self.assertEqual(q.wait(), 'a')
sleep(0) sleep(0)
self.assertEquals(results, ['a', 'b', 'c']) self.assertEqual(results, ['a', 'b', 'c'])
self.assertEquals(q.wait(), 'b') self.assertEqual(q.wait(), 'b')
self.assertEquals(q.wait(), 'c') self.assertEqual(q.wait(), 'c')
@silence_warnings @silence_warnings
def test_zero_max_size(self): def test_zero_max_size(self):
@@ -72,8 +72,8 @@ class TestQueue(LimitedTestCase):
sleep(0) sleep(0)
self.assert_(not e1.ready()) self.assert_(not e1.ready())
spawn(receiver, e2, q) spawn(receiver, e2, q)
self.assertEquals(e2.wait(),'hi') self.assertEqual(e2.wait(),'hi')
self.assertEquals(e1.wait(),'done') self.assertEqual(e1.wait(),'done')
@silence_warnings @silence_warnings
def test_multiple_waiters(self): def test_multiple_waiters(self):
@@ -93,7 +93,7 @@ class TestQueue(LimitedTestCase):
results = set() results = set()
for i, gt in enumerate(gts): for i, gt in enumerate(gts):
results.add(gt.wait()) results.add(gt.wait())
self.assertEquals(results, set(sendings)) self.assertEqual(results, set(sendings))
@silence_warnings @silence_warnings
def test_waiters_that_cancel(self): def test_waiters_that_cancel(self):
@@ -110,10 +110,10 @@ class TestQueue(LimitedTestCase):
evt = Event() evt = Event()
spawn(do_receive, q, evt) spawn(do_receive, q, evt)
self.assertEquals(evt.wait(), 'timed out') self.assertEqual(evt.wait(), 'timed out')
q.send('hi') q.send('hi')
self.assertEquals(q.wait(), 'hi') self.assertEqual(q.wait(), 'hi')
@silence_warnings @silence_warnings
def test_senders_that_die(self): def test_senders_that_die(self):
@@ -123,7 +123,7 @@ class TestQueue(LimitedTestCase):
q.send('sent') q.send('sent')
spawn(do_send, q) spawn(do_send, q)
self.assertEquals(q.wait(), 'sent') self.assertEqual(q.wait(), 'sent')
@silence_warnings @silence_warnings
def test_two_waiters_one_dies(self): def test_two_waiters_one_dies(self):
@@ -144,8 +144,8 @@ class TestQueue(LimitedTestCase):
spawn(waiter, q, waiting_evt) spawn(waiter, q, waiting_evt)
sleep(0) sleep(0)
q.send('hi') q.send('hi')
self.assertEquals(dying_evt.wait(), 'timed out') self.assertEqual(dying_evt.wait(), 'timed out')
self.assertEquals(waiting_evt.wait(), 'hi') self.assertEqual(waiting_evt.wait(), 'hi')
@silence_warnings @silence_warnings
def test_two_bogus_waiters(self): def test_two_bogus_waiters(self):
@@ -164,9 +164,9 @@ class TestQueue(LimitedTestCase):
spawn(do_receive, q, e2) spawn(do_receive, q, e2)
sleep(0) sleep(0)
q.send('sent') q.send('sent')
self.assertEquals(e1.wait(), 'timed out') self.assertEqual(e1.wait(), 'timed out')
self.assertEquals(e2.wait(), 'timed out') self.assertEqual(e2.wait(), 'timed out')
self.assertEquals(q.wait(), 'sent') self.assertEqual(q.wait(), 'sent')
@silence_warnings @silence_warnings
def test_waiting(self): def test_waiting(self):
@@ -178,12 +178,12 @@ class TestQueue(LimitedTestCase):
e1 = Event() e1 = Event()
spawn(do_wait, q, e1) spawn(do_wait, q, e1)
sleep(0) sleep(0)
self.assertEquals(1, q.waiting()) self.assertEqual(1, q.waiting())
q.send('hi') q.send('hi')
sleep(0) sleep(0)
self.assertEquals(0, q.waiting()) self.assertEqual(0, q.waiting())
self.assertEquals('hi', e1.wait()) self.assertEqual('hi', e1.wait())
self.assertEquals(0, q.waiting()) self.assertEqual(0, q.waiting())
class TestChannel(LimitedTestCase): class TestChannel(LimitedTestCase):
@@ -240,19 +240,19 @@ class TestChannel(LimitedTestCase):
w2 = eventlet.spawn(c.wait) w2 = eventlet.spawn(c.wait)
w3 = eventlet.spawn(c.wait) w3 = eventlet.spawn(c.wait)
sleep(0) sleep(0)
self.assertEquals(c.waiting(), 3) self.assertEqual(c.waiting(), 3)
s1 = eventlet.spawn(c.send, 1) s1 = eventlet.spawn(c.send, 1)
s2 = eventlet.spawn(c.send, 2) s2 = eventlet.spawn(c.send, 2)
s3 = eventlet.spawn(c.send, 3) s3 = eventlet.spawn(c.send, 3)
sleep(0) # this gets all the sends into a waiting state sleep(0) # this gets all the sends into a waiting state
self.assertEquals(c.waiting(), 0) self.assertEqual(c.waiting(), 0)
s1.wait() s1.wait()
s2.wait() s2.wait()
s3.wait() s3.wait()
# NOTE: we don't guarantee that waiters are served in order # NOTE: we don't guarantee that waiters are served in order
results = sorted([w1.wait(), w2.wait(), w3.wait()]) results = sorted([w1.wait(), w2.wait(), w3.wait()])
self.assertEquals(results, [1,2,3]) self.assertEqual(results, [1,2,3])
if __name__=='__main__': if __name__=='__main__':
main() main()

18
tests/test__pool.py
View File

@@ -68,7 +68,7 @@ class TestCoroutinePool(LimitedTestCase):
done = pool.execute(consumer) done = pool.execute(consumer)
pool.execute_async(producer) pool.execute_async(producer)
done.wait() done.wait()
self.assertEquals(['cons1', 'prod', 'cons2'], results) self.assertEqual(['cons1', 'prod', 'cons2'], results)
def test_timer_cancel(self): def test_timer_cancel(self):
# this test verifies that local timers are not fired # this test verifies that local timers are not fired
@@ -82,7 +82,7 @@ class TestCoroutinePool(LimitedTestCase):
worker = pool.execute(some_work) worker = pool.execute(some_work)
worker.wait() worker.wait()
api.sleep(0) api.sleep(0)
self.assertEquals(timer_fired, []) self.assertEqual(timer_fired, [])
def test_reentrant(self): def test_reentrant(self):
pool = self.klass(0,1) pool = self.klass(0,1)
@@ -131,7 +131,7 @@ class TestCoroutinePool(LimitedTestCase):
e.wait() e.wait()
pool.execute(wait_long_time, evt) pool.execute(wait_long_time, evt)
pool.execute(wait_long_time, evt) pool.execute(wait_long_time, evt)
self.assertEquals(pool.free(), 0) self.assertEqual(pool.free(), 0)
self.assert_pool_has_free(pool, 0) self.assert_pool_has_free(pool, 0)
# verify that the pool discards excess items put into it # verify that the pool discards excess items put into it
@@ -143,12 +143,12 @@ class TestCoroutinePool(LimitedTestCase):
api.sleep(0) api.sleep(0)
api.sleep(0) api.sleep(0)
self.assertEquals(pool.free(), 1) self.assertEqual(pool.free(), 1)
self.assert_pool_has_free(pool, 1) self.assert_pool_has_free(pool, 1)
# resize larger and assert that there are more free items # resize larger and assert that there are more free items
pool.resize(2) pool.resize(2)
self.assertEquals(pool.free(), 2) self.assertEqual(pool.free(), 2)
self.assert_pool_has_free(pool, 2) self.assert_pool_has_free(pool, 2)
def test_stderr_raising(self): def test_stderr_raising(self):
@@ -199,7 +199,7 @@ class TestCoroutinePool(LimitedTestCase):
api.sleep(0.1) api.sleep(0.1)
return 'ok' return 'ok'
pool.execute(slow) pool.execute(slow)
self.assertEquals(pool.wait(), 'ok') self.assertEqual(pool.wait(), 'ok')
def test_pool_smash(self): def test_pool_smash(self):
# The premise is that a coroutine in a Pool tries to get a token out # The premise is that a coroutine in a Pool tries to get a token out
@@ -220,7 +220,7 @@ class TestCoroutinePool(LimitedTestCase):
# the execute makes the token pool expect that coroutine, but then # the execute makes the token pool expect that coroutine, but then
# immediately cuts bait # immediately cuts bait
e1 = pool.execute(do_receive, tp) e1 = pool.execute(do_receive, tp)
self.assertEquals(e1.wait(), 'timed out') self.assertEqual(e1.wait(), 'timed out')
# the pool can get some random item back # the pool can get some random item back
def send_wakeup(tp): def send_wakeup(tp):
@@ -232,10 +232,10 @@ class TestCoroutinePool(LimitedTestCase):
def resume(): def resume():
return 'resumed' return 'resumed'
e2 = pool.execute(resume) e2 = pool.execute(resume)
self.assertEquals(e2.wait(), 'resumed') self.assertEqual(e2.wait(), 'resumed')
# we should be able to get out the thing we put in there, too # we should be able to get out the thing we put in there, too
self.assertEquals(tp.get(), 'wakeup') self.assertEqual(tp.get(), 'wakeup')
class PoolBasicTests(LimitedTestCase): class PoolBasicTests(LimitedTestCase):

6
tests/thread_test.py
View File

@@ -84,16 +84,16 @@ class Locals(LimitedTestCase):
my_local = Caller() my_local = Caller()
my_local.foo = "foo1" my_local.foo = "foo1"
self.assertEquals("foo1", my_local.callme()) self.assertEqual("foo1", my_local.callme())
def do_something(): def do_something():
my_local.foo = "foo2" my_local.foo = "foo2"
self.assertEquals("foo2", my_local.callme()) self.assertEqual("foo2", my_local.callme())
eventlet.spawn(do_something).wait() eventlet.spawn(do_something).wait()
my_local.foo = "foo3" my_local.foo = "foo3"
self.assertEquals("foo3", my_local.callme()) self.assertEqual("foo3", my_local.callme())
def test_no_leaking(self): def test_no_leaking(self):
refs = weakref.WeakKeyDictionary() refs = weakref.WeakKeyDictionary()

78
tests/tpool_test.py
View File

@@ -14,17 +14,15 @@
# limitations under the License. # limitations under the License.
from __future__ import print_function from __future__ import print_function
import itertools
import random
from sys import stdout
import time
import re
import gc import gc
from tests import skipped, skip_with_pyevent, LimitedTestCase, main import random
import re
import time
from eventlet import tpool, debug
from eventlet.support import six
import eventlet import eventlet
from eventlet import tpool
from eventlet.support import six
from tests import LimitedTestCase, skipped, skip_with_pyevent, main
one = 1 one = 1
@@ -68,8 +66,10 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_wrap_uniterable(self): def test_wrap_uniterable(self):
prox = tpool.Proxy([]) prox = tpool.Proxy([])
def index(): def index():
prox[0] prox[0]
def key(): def key():
prox['a'] prox['a']
@@ -78,7 +78,7 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_wrap_dict(self): def test_wrap_dict(self):
my_object = {'a':1} my_object = {'a': 1}
prox = tpool.Proxy(my_object) prox = tpool.Proxy(my_object)
self.assertEqual('a', prox.keys()[0]) self.assertEqual('a', prox.keys()[0])
self.assertEqual(1, prox['a']) self.assertEqual(1, prox['a'])
@@ -111,9 +111,9 @@ class TestTpool(LimitedTestCase):
def test_wrap_hash(self): def test_wrap_hash(self):
prox1 = tpool.Proxy(''+'A') prox1 = tpool.Proxy(''+'A')
prox2 = tpool.Proxy('A'+'') prox2 = tpool.Proxy('A'+'')
self.assert_(prox1=='A') self.assert_(prox1 == 'A')
self.assert_('A'==prox2) self.assert_('A' == prox2)
#self.assert_(prox1==prox2) FIXME - could __eq__ unwrap rhs if it is other proxy? #self.assert_(prox1 == prox2) FIXME - could __eq__ unwrap rhs if it is other proxy?
self.assertEqual(hash(prox1), hash(prox2)) self.assertEqual(hash(prox1), hash(prox2))
proxList = tpool.Proxy([]) proxList = tpool.Proxy([])
self.assertRaises(TypeError, hash, proxList) self.assertRaises(TypeError, hash, proxList)
@@ -130,19 +130,19 @@ class TestTpool(LimitedTestCase):
def test_multiple_wraps(self): def test_multiple_wraps(self):
prox1 = tpool.Proxy(re) prox1 = tpool.Proxy(re)
prox2 = tpool.Proxy(re) prox2 = tpool.Proxy(re)
x1 = prox1.compile('.') prox1.compile('.')
x2 = prox1.compile('.') x2 = prox1.compile('.')
del x2 del x2
x3 = prox2.compile('.') prox2.compile('.')
@skip_with_pyevent @skip_with_pyevent
def test_wrap_getitem(self): def test_wrap_getitem(self):
prox = tpool.Proxy([0,1,2]) prox = tpool.Proxy([0, 1, 2])
self.assertEqual(prox[0], 0) self.assertEqual(prox[0], 0)
@skip_with_pyevent @skip_with_pyevent
def test_wrap_setitem(self): def test_wrap_setitem(self):
prox = tpool.Proxy([0,1,2]) prox = tpool.Proxy([0, 1, 2])
prox[1] = 2 prox[1] = 2
self.assertEqual(prox[1], 2) self.assertEqual(prox[1], 2)
@@ -153,11 +153,12 @@ class TestTpool(LimitedTestCase):
result = [] result = []
for i in prox: for i in prox:
result.append(i) result.append(i)
self.assertEquals(list(range(10)), result) self.assertEqual(list(range(10)), result)
@skip_with_pyevent @skip_with_pyevent
def test_wrap_iterator2(self): def test_wrap_iterator2(self):
self.reset_timeout(5) # might take a while due to imprecise sleeping self.reset_timeout(5) # might take a while due to imprecise sleeping
def foo(): def foo():
import time import time
for x in range(2): for x in range(2):
@@ -165,9 +166,10 @@ class TestTpool(LimitedTestCase):
time.sleep(0.001) time.sleep(0.001)
counter = [0] counter = [0]
def tick(): def tick():
for i in six.moves.range(20000): for i in six.moves.range(20000):
counter[0]+=1 counter[0] += 1
if counter[0] % 20 == 0: if counter[0] % 20 == 0:
eventlet.sleep(0.0001) eventlet.sleep(0.0001)
else: else:
@@ -185,6 +187,7 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_raising_exceptions(self): def test_raising_exceptions(self):
prox = tpool.Proxy(re) prox = tpool.Proxy(re)
def nofunc(): def nofunc():
prox.never_name_a_function_like_this() prox.never_name_a_function_like_this()
self.assertRaises(AttributeError, nofunc) self.assertRaises(AttributeError, nofunc)
@@ -197,8 +200,8 @@ class TestTpool(LimitedTestCase):
def test_variable_and_keyword_arguments_with_function_calls(self): def test_variable_and_keyword_arguments_with_function_calls(self):
import optparse import optparse
parser = tpool.Proxy(optparse.OptionParser()) parser = tpool.Proxy(optparse.OptionParser())
z = parser.add_option('-n', action='store', type='string', dest='n') parser.add_option('-n', action='store', type='string', dest='n')
opts,args = parser.parse_args(["-nfoo"]) opts, args = parser.parse_args(["-nfoo"])
self.assertEqual(opts.n, 'foo') self.assertEqual(opts.n, 'foo')
@skip_with_pyevent @skip_with_pyevent
@@ -207,11 +210,11 @@ class TestTpool(LimitedTestCase):
prox = tpool.Proxy(tpool_test) prox = tpool.Proxy(tpool_test)
pile = eventlet.GreenPile(4) pile = eventlet.GreenPile(4)
pile.spawn(lambda: self.assertEquals(prox.one, 1)) pile.spawn(lambda: self.assertEqual(prox.one, 1))
pile.spawn(lambda: self.assertEquals(prox.two, 2)) pile.spawn(lambda: self.assertEqual(prox.two, 2))
pile.spawn(lambda: self.assertEquals(prox.three, 3)) pile.spawn(lambda: self.assertEqual(prox.three, 3))
results = list(pile) results = list(pile)
self.assertEquals(len(results), 3) self.assertEqual(len(results), 3)
@skip_with_pyevent @skip_with_pyevent
def test_timeout(self): def test_timeout(self):
@@ -227,7 +230,7 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_autowrap(self): def test_autowrap(self):
x = tpool.Proxy({'a':1, 'b':2}, autowrap=(int,)) x = tpool.Proxy({'a': 1, 'b': 2}, autowrap=(int,))
self.assert_(isinstance(x.get('a'), tpool.Proxy)) self.assert_(isinstance(x.get('a'), tpool.Proxy))
self.assert_(not isinstance(x.items(), tpool.Proxy)) self.assert_(not isinstance(x.items(), tpool.Proxy))
# attributes as well as callables # attributes as well as callables
@@ -238,7 +241,7 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_autowrap_names(self): def test_autowrap_names(self):
x = tpool.Proxy({'a':1, 'b':2}, autowrap_names=('get',)) x = tpool.Proxy({'a': 1, 'b': 2}, autowrap_names=('get',))
self.assert_(isinstance(x.get('a'), tpool.Proxy)) self.assert_(isinstance(x.get('a'), tpool.Proxy))
self.assert_(not isinstance(x.items(), tpool.Proxy)) self.assert_(not isinstance(x.items(), tpool.Proxy))
from tests import tpool_test from tests import tpool_test
@@ -259,11 +262,11 @@ class TestTpool(LimitedTestCase):
def wrapped(arg): def wrapped(arg):
return arg return arg
x = tpool.Proxy(wrapped) x = tpool.Proxy(wrapped)
self.assertEquals(4, x(4)) self.assertEqual(4, x(4))
# verify that it wraps return values if specified # verify that it wraps return values if specified
x = tpool.Proxy(wrapped, autowrap_names=('__call__',)) x = tpool.Proxy(wrapped, autowrap_names=('__call__',))
self.assert_(isinstance(x(4), tpool.Proxy)) self.assert_(isinstance(x(4), tpool.Proxy))
self.assertEquals("4", str(x(4))) self.assertEqual("4", str(x(4)))
@skip_with_pyevent @skip_with_pyevent
def test_callable_iterator(self): def test_callable_iterator(self):
@@ -274,7 +277,7 @@ class TestTpool(LimitedTestCase):
x = tpool.Proxy(wrapped, autowrap_names=('__call__',)) x = tpool.Proxy(wrapped, autowrap_names=('__call__',))
for r in x(3): for r in x(3):
self.assertEquals(3, r) self.assertEqual(3, r)
@skip_with_pyevent @skip_with_pyevent
def test_eventlet_timeout(self): def test_eventlet_timeout(self):
@@ -285,15 +288,18 @@ class TestTpool(LimitedTestCase):
@skip_with_pyevent @skip_with_pyevent
def test_tpool_set_num_threads(self): def test_tpool_set_num_threads(self):
tpool.set_num_threads(5) tpool.set_num_threads(5)
self.assertEquals(5, tpool._nthreads) self.assertEqual(5, tpool._nthreads)
class TpoolLongTests(LimitedTestCase): class TpoolLongTests(LimitedTestCase):
TEST_TIMEOUT=60 TEST_TIMEOUT = 60
@skip_with_pyevent @skip_with_pyevent
def test_a_buncha_stuff(self): def test_a_buncha_stuff(self):
assert_ = self.assert_ assert_ = self.assert_
class Dummy(object): class Dummy(object):
def foo(self,when,token=None): def foo(self, when, token=None):
assert_(token is not None) assert_(token is not None)
time.sleep(random.random()/200.0) time.sleep(random.random()/200.0)
return token return token
@@ -305,16 +311,16 @@ class TpoolLongTests(LimitedTestCase):
eventlet.sleep(random.random()/200.0) eventlet.sleep(random.random()/200.0)
now = time.time() now = time.time()
token = loopnum * count + n token = loopnum * count + n
rv = obj.foo(now,token=token) rv = obj.foo(now, token=token)
self.assertEquals(token, rv) self.assertEqual(token, rv)
eventlet.sleep(random.random()/200.0) eventlet.sleep(random.random()/200.0)
cnt = 10 cnt = 10
pile = eventlet.GreenPile(cnt) pile = eventlet.GreenPile(cnt)
for i in six.moves.range(cnt): for i in six.moves.range(cnt):
pile.spawn(sender_loop,i) pile.spawn(sender_loop, i)
results = list(pile) results = list(pile)
self.assertEquals(len(results), cnt) self.assertEqual(len(results), cnt)
tpool.killall() tpool.killall()
@skipped @skipped

58
tests/wsgi_test.py
View File

@@ -447,7 +447,7 @@ class TestHttpd(_TestBase):
sock = eventlet.wrap_ssl(sock) 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('POST /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\nContent-length:3\r\n\r\nabc')
result = sock.read(8192) result = sock.read(8192)
self.assertEquals(result[-3:], 'abc') self.assertEqual(result[-3:], 'abc')
@skip_if_no_ssl @skip_if_no_ssl
def test_013_empty_return(self): def test_013_empty_return(self):
@@ -467,7 +467,7 @@ class TestHttpd(_TestBase):
sock = eventlet.wrap_ssl(sock) sock = eventlet.wrap_ssl(sock)
sock.write('GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n') sock.write('GET /foo HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n')
result = sock.read(8192) result = sock.read(8192)
self.assertEquals(result[-4:], '\r\n\r\n') self.assertEqual(result[-4:], '\r\n\r\n')
def test_014_chunked_post(self): def test_014_chunked_post(self):
self.site.application = chunked_post self.site.application = chunked_post
@@ -544,7 +544,7 @@ class TestHttpd(_TestBase):
break break
else: else:
header_lines.append(line) header_lines.append(line)
self.assertEquals(1, len( self.assertEqual(1, len(
[l for l in header_lines if l.lower().startswith('content-length')])) [l for l in header_lines if l.lower().startswith('content-length')]))
@skip_if_no_ssl @skip_if_no_ssl
@@ -748,8 +748,8 @@ class TestHttpd(_TestBase):
fd.write('PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 1025\r\nExpect: 100-continue\r\n\r\n') fd.write('PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 1025\r\nExpect: 100-continue\r\n\r\n')
fd.flush() fd.flush()
result = read_http(sock) result = read_http(sock)
self.assertEquals(result.status, 'HTTP/1.1 417 Expectation Failed') self.assertEqual(result.status, 'HTTP/1.1 417 Expectation Failed')
self.assertEquals(result.body, 'failure') 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('PUT / HTTP/1.1\r\nHost: localhost\r\nContent-length: 7\r\nExpect: 100-continue\r\n\r\ntesting')
fd.flush() fd.flush()
header_lines = [] header_lines = []
@@ -768,7 +768,7 @@ class TestHttpd(_TestBase):
else: else:
header_lines.append(line) header_lines.append(line)
self.assert_(header_lines[0].startswith('HTTP/1.1 200 OK')) self.assert_(header_lines[0].startswith('HTTP/1.1 200 OK'))
self.assertEquals(fd.read(7), 'testing') self.assertEqual(fd.read(7), 'testing')
fd.close() fd.close()
sock.close() sock.close()
@@ -787,7 +787,7 @@ class TestHttpd(_TestBase):
eventlet.connect(('localhost', self.port)) eventlet.connect(('localhost', self.port))
self.fail("Didn't expect to connect") self.fail("Didn't expect to connect")
except socket.error as exc: except socket.error as exc:
self.assertEquals(get_errno(exc), errno.ECONNREFUSED) self.assertEqual(get_errno(exc), errno.ECONNREFUSED)
self.assert_('Invalid argument' in self.logfile.getvalue(), self.assert_('Invalid argument' in self.logfile.getvalue(),
self.logfile.getvalue()) self.logfile.getvalue())
@@ -815,7 +815,7 @@ class TestHttpd(_TestBase):
result = read_http(sock) result = read_http(sock)
self.assertEqual(result.headers_lower['connection'], 'close') self.assertEqual(result.headers_lower['connection'], 'close')
self.assertNotEqual(result.headers_lower.get('transfer-encoding'), 'chunked') self.assertNotEqual(result.headers_lower.get('transfer-encoding'), 'chunked')
self.assertEquals(result.body, "thisischunked") self.assertEqual(result.body, "thisischunked")
def test_minimum_chunk_size_parameter_leaves_httpprotocol_class_member_intact(self): def test_minimum_chunk_size_parameter_leaves_httpprotocol_class_member_intact(self):
start_size = wsgi.HttpProtocol.minimum_chunk_size start_size = wsgi.HttpProtocol.minimum_chunk_size
@@ -922,11 +922,11 @@ class TestHttpd(_TestBase):
posthook2_count = [0] posthook2_count = [0]
def posthook1(env, value, multiplier=1): def posthook1(env, value, multiplier=1):
self.assertEquals(env['local.test'], 'test_029_posthooks') self.assertEqual(env['local.test'], 'test_029_posthooks')
posthook1_count[0] += value * multiplier posthook1_count[0] += value * multiplier
def posthook2(env, value, divisor=1): def posthook2(env, value, divisor=1):
self.assertEquals(env['local.test'], 'test_029_posthooks') self.assertEqual(env['local.test'], 'test_029_posthooks')
posthook2_count[0] += value / divisor posthook2_count[0] += value / divisor
def one_posthook_app(env, start_response): def one_posthook_app(env, start_response):
@@ -944,11 +944,11 @@ class TestHttpd(_TestBase):
fp = sock.makefile('rw') fp = sock.makefile('rw')
fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n') fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.flush() fp.flush()
self.assertEquals(fp.readline(), 'HTTP/1.1 200 OK\r\n') self.assertEqual(fp.readline(), 'HTTP/1.1 200 OK\r\n')
fp.close() fp.close()
sock.close() sock.close()
self.assertEquals(posthook1_count[0], 6) self.assertEqual(posthook1_count[0], 6)
self.assertEquals(posthook2_count[0], 0) self.assertEqual(posthook2_count[0], 0)
def two_posthook_app(env, start_response): def two_posthook_app(env, start_response):
env['local.test'] = 'test_029_posthooks' env['local.test'] = 'test_029_posthooks'
@@ -967,11 +967,11 @@ class TestHttpd(_TestBase):
fp = sock.makefile('rw') fp = sock.makefile('rw')
fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n') fp.write('GET / HTTP/1.1\r\nHost: localhost\r\n\r\n')
fp.flush() fp.flush()
self.assertEquals(fp.readline(), 'HTTP/1.1 200 OK\r\n') self.assertEqual(fp.readline(), 'HTTP/1.1 200 OK\r\n')
fp.close() fp.close()
sock.close() sock.close()
self.assertEquals(posthook1_count[0], 26) self.assertEqual(posthook1_count[0], 26)
self.assertEquals(posthook2_count[0], 25) self.assertEqual(posthook2_count[0], 25)
def test_030_reject_long_header_lines(self): def test_030_reject_long_header_lines(self):
sock = eventlet.connect(('localhost', self.port)) sock = eventlet.connect(('localhost', self.port))
@@ -981,7 +981,7 @@ class TestHttpd(_TestBase):
fd.write(request) fd.write(request)
fd.flush() fd.flush()
result = read_http(sock) result = read_http(sock)
self.assertEquals(result.status, 'HTTP/1.0 400 Header Line Too Long') self.assertEqual(result.status, 'HTTP/1.0 400 Header Line Too Long')
fd.close() fd.close()
def test_031_reject_large_headers(self): def test_031_reject_large_headers(self):
@@ -992,7 +992,7 @@ class TestHttpd(_TestBase):
fd.write(request) fd.write(request)
fd.flush() fd.flush()
result = read_http(sock) result = read_http(sock)
self.assertEquals(result.status, 'HTTP/1.0 400 Headers Too Large') self.assertEqual(result.status, 'HTTP/1.0 400 Headers Too Large')
fd.close() fd.close()
def test_032_wsgi_input_as_iterable(self): def test_032_wsgi_input_as_iterable(self):
@@ -1019,9 +1019,9 @@ class TestHttpd(_TestBase):
fd.write(request) fd.write(request)
fd.flush() fd.flush()
result = read_http(sock) result = read_http(sock)
self.assertEquals(result.body, upload_data) self.assertEqual(result.body, upload_data)
fd.close() fd.close()
self.assertEquals(g[0], 1) self.assertEqual(g[0], 1)
def test_zero_length_chunked_response(self): def test_zero_length_chunked_response(self):
def zero_chunked_app(env, start_response): def zero_chunked_app(env, start_response):
@@ -1400,7 +1400,7 @@ class TestChunkedInput(_TestBase):
def ping(self, fd): def ping(self, fd):
fd.sendall("GET /ping HTTP/1.1\r\n\r\n") fd.sendall("GET /ping HTTP/1.1\r\n\r\n")
self.assertEquals(read_http(fd).body, "pong") self.assertEqual(read_http(fd).body, "pong")
def test_short_read_with_content_length(self): def test_short_read_with_content_length(self):
body = self.body() body = self.body()
@@ -1408,7 +1408,7 @@ class TestChunkedInput(_TestBase):
fd = self.connect() fd = self.connect()
fd.sendall(req) fd.sendall(req)
self.assertEquals(read_http(fd).body, "this is ch") self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd) self.ping(fd)
fd.close() fd.close()
@@ -1418,7 +1418,7 @@ class TestChunkedInput(_TestBase):
req = "POST /short-read HTTP/1.1\r\ntransfer-encoding: Chunked\r\nContent-Length:0\r\n\r\n" + 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 = self.connect()
fd.sendall(req) fd.sendall(req)
self.assertEquals(read_http(fd).body, "this is ch") self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd) self.ping(fd)
fd.close() fd.close()
@@ -1429,7 +1429,7 @@ class TestChunkedInput(_TestBase):
fd = self.connect() fd = self.connect()
fd.sendall(req) fd.sendall(req)
self.assertEquals(read_http(fd).body, "this is ch") self.assertEqual(read_http(fd).body, "this is ch")
self.ping(fd) self.ping(fd)
fd.close() fd.close()
@@ -1440,7 +1440,7 @@ class TestChunkedInput(_TestBase):
fd = self.connect() fd = self.connect()
fd.sendall(req) fd.sendall(req)
self.assertEquals(read_http(fd).body, "pong") self.assertEqual(read_http(fd).body, "pong")
self.ping(fd) self.ping(fd)
fd.close() fd.close()
@@ -1451,7 +1451,7 @@ class TestChunkedInput(_TestBase):
fd = self.connect() fd = self.connect()
fd.sendall(req) fd.sendall(req)
self.assertEquals(read_http(fd).body, 'this is chunked\nline 2\nline3') self.assertEqual(read_http(fd).body, 'this is chunked\nline 2\nline3')
fd.close() fd.close()
def test_chunked_readline_wsgi_override_minimum_chunk_size(self): def test_chunked_readline_wsgi_override_minimum_chunk_size(self):
@@ -1466,7 +1466,7 @@ class TestChunkedInput(_TestBase):
resp_so_far += one_byte resp_so_far += one_byte
if resp_so_far.endswith('\r\n\r\n'): if resp_so_far.endswith('\r\n\r\n'):
break break
self.assertEquals(fd.recv(1), ' ') self.assertEqual(fd.recv(1), ' ')
try: try:
with eventlet.Timeout(.1): with eventlet.Timeout(.1):
fd.recv(1) fd.recv(1)
@@ -1477,7 +1477,7 @@ class TestChunkedInput(_TestBase):
self.yield_next_space = True self.yield_next_space = True
with eventlet.Timeout(.1): with eventlet.Timeout(.1):
self.assertEquals(fd.recv(1), ' ') self.assertEqual(fd.recv(1), ' ')
def test_chunked_readline_wsgi_not_override_minimum_chunk_size(self): def test_chunked_readline_wsgi_not_override_minimum_chunk_size(self):
@@ -1492,7 +1492,7 @@ class TestChunkedInput(_TestBase):
resp_so_far += one_byte resp_so_far += one_byte
if resp_so_far.endswith('\r\n\r\n'): if resp_so_far.endswith('\r\n\r\n'):
break break
self.assertEquals(fd.recv(1), ' ') self.assertEqual(fd.recv(1), ' ')
except eventlet.Timeout: except eventlet.Timeout:
pass pass
else: else:

23
tests/zmq_test.py
View File

@@ -1,12 +1,9 @@
from __future__ import with_statement from __future__ import with_statement
from eventlet import event, spawn, sleep, patcher, semaphore from eventlet import event, spawn, sleep, semaphore
from eventlet.hubs import get_hub, _threadlocal, use_hub
from nose.tools import * from nose.tools import *
from tests import check_idle_cpu_usage, mock, LimitedTestCase, using_pyevent, skip_unless from tests import check_idle_cpu_usage, LimitedTestCase, using_pyevent, skip_unless
from unittest import TestCase
from threading import Thread
try: try:
from eventlet.green import zmq from eventlet.green import zmq
except ImportError: except ImportError:
@@ -197,7 +194,7 @@ class TestUpstreamDownStream(LimitedTestCase):
def tx(sock): def tx(sock):
for i in range(1, 1001): for i in range(1, 1001):
msg = "sub%s %s" % ([2,1][i % 2], i) msg = "sub%s %s" % ([2, 1][i % 2], i)
sock.send(msg) sock.send(msg)
sleep() sleep()
sock.send('sub1 LAST') sock.send('sub1 LAST')
@@ -312,7 +309,6 @@ class TestUpstreamDownStream(LimitedTestCase):
for evt in done_evts: for evt in done_evts:
self.assertEqual(evt.wait(), 0) self.assertEqual(evt.wait(), 0)
@skip_unless(zmq_supported) @skip_unless(zmq_supported)
def test_send_during_recv_multipart(self): def test_send_during_recv_multipart(self):
sender, receiver, port = self.create_bound_pair(zmq.XREQ, zmq.XREQ) sender, receiver, port = self.create_bound_pair(zmq.XREQ, zmq.XREQ)
@@ -349,14 +345,12 @@ class TestUpstreamDownStream(LimitedTestCase):
final_i = done_evts[i].wait() final_i = done_evts[i].wait()
self.assertEqual(final_i, 0) self.assertEqual(final_i, 0)
# Need someway to ensure a thread is blocked on send... This isn't working # Need someway to ensure a thread is blocked on send... This isn't working
@skip_unless(zmq_supported) @skip_unless(zmq_supported)
def test_recv_during_send(self): def test_recv_during_send(self):
sender, receiver, port = self.create_bound_pair(zmq.XREQ, zmq.XREQ) sender, receiver, port = self.create_bound_pair(zmq.XREQ, zmq.XREQ)
sleep() sleep()
num_recvs = 30
done = event.Event() done = event.Event()
try: try:
@@ -480,12 +474,12 @@ class TestQueueLock(LimitedTestCase):
spawn(lock, 3) spawn(lock, 3)
sleep() sleep()
self.assertEquals(results, []) self.assertEqual(results, [])
q.release() q.release()
s.acquire() s.acquire()
s.acquire() s.acquire()
s.acquire() s.acquire()
self.assertEquals(results, [1,2,3]) self.assertEqual(results, [1, 2, 3])
@skip_unless(zmq_supported) @skip_unless(zmq_supported)
def test_count(self): def test_count(self):
@@ -520,6 +514,7 @@ class TestQueueLock(LimitedTestCase):
s = semaphore.Semaphore(0) s = semaphore.Semaphore(0)
results = [] results = []
def lock(x): def lock(x):
with q: with q:
results.append(x) results.append(x)
@@ -527,15 +522,15 @@ class TestQueueLock(LimitedTestCase):
spawn(lock, 1) spawn(lock, 1)
sleep() sleep()
self.assertEquals(results, []) self.assertEqual(results, [])
q.release() q.release()
sleep() sleep()
self.assertEquals(results, []) self.assertEqual(results, [])
self.assertTrue(q) self.assertTrue(q)
q.release() q.release()
s.acquire() s.acquire()
self.assertEquals(results, [1]) self.assertEqual(results, [1])
class TestBlockedThread(LimitedTestCase): class TestBlockedThread(LimitedTestCase):