Skipping repeated selects test on zmq hub because it doesn't appear to support detecting bad filenos at select() time.
This commit is contained in:
@@ -88,6 +88,23 @@ def using_pyevent(_f):
|
|||||||
from eventlet.hubs import get_hub
|
from eventlet.hubs import get_hub
|
||||||
return 'pyevent' in type(get_hub()).__module__
|
return 'pyevent' in type(get_hub()).__module__
|
||||||
|
|
||||||
|
|
||||||
|
def using_zmq(_f):
|
||||||
|
try:
|
||||||
|
import zmq
|
||||||
|
except ImportError:
|
||||||
|
return False
|
||||||
|
from eventlet.hubs import get_hub
|
||||||
|
return zmq and 'zeromq' in type(get_hub()).__module__
|
||||||
|
|
||||||
|
def skip_unless_zmq(func):
|
||||||
|
""" Decorator that skips a test if we're not using the zeromq hub."""
|
||||||
|
return skip_unless(using_zmq)(func)
|
||||||
|
|
||||||
|
def skip_with_zmq(func):
|
||||||
|
""" Decorator that skips a test if we're using the zeromq hub."""
|
||||||
|
return skip_if(using_zmq)(func)
|
||||||
|
|
||||||
|
|
||||||
def skip_with_pyevent(func):
|
def skip_with_pyevent(func):
|
||||||
""" Decorator that skips a test if we're using the pyevent hub."""
|
""" Decorator that skips a test if we're using the pyevent hub."""
|
||||||
|
@@ -1,4 +1,4 @@
|
|||||||
from tests import LimitedTestCase, main, skip_with_pyevent, skip_if_no_itimer
|
from tests import LimitedTestCase, main, skip_with_pyevent, skip_if_no_itimer, skip_with_zmq
|
||||||
import time
|
import time
|
||||||
import eventlet
|
import eventlet
|
||||||
from eventlet import hubs
|
from eventlet import hubs
|
||||||
@@ -208,6 +208,7 @@ except eventlet.Timeout:
|
|||||||
|
|
||||||
class TestBadFilenos(LimitedTestCase):
|
class TestBadFilenos(LimitedTestCase):
|
||||||
@skip_with_pyevent
|
@skip_with_pyevent
|
||||||
|
@skip_with_zmq
|
||||||
def test_repeated_selects(self):
|
def test_repeated_selects(self):
|
||||||
from eventlet.green import select
|
from eventlet.green import select
|
||||||
self.assertRaises(ValueError, select.select, [-1], [], [])
|
self.assertRaises(ValueError, select.select, [-1], [], [])
|
||||||
|
@@ -1,7 +1,7 @@
|
|||||||
from eventlet import event, spawn, sleep, patcher
|
from eventlet import event, spawn, sleep, patcher
|
||||||
from eventlet.hubs import get_hub, _threadlocal, use_hub
|
from eventlet.hubs import get_hub, _threadlocal, use_hub
|
||||||
from nose.tools import *
|
from nose.tools import *
|
||||||
from tests import mock, LimitedTestCase, skip_unless
|
from tests import mock, LimitedTestCase, skip_unless_zmq
|
||||||
from unittest import TestCase
|
from unittest import TestCase
|
||||||
|
|
||||||
from threading import Thread
|
from threading import Thread
|
||||||
@@ -12,13 +12,6 @@ except ImportError:
|
|||||||
zmq = None
|
zmq = None
|
||||||
Hub = None
|
Hub = None
|
||||||
|
|
||||||
def using_zmq(_f):
|
|
||||||
return zmq and 'zeromq' in type(get_hub()).__module__
|
|
||||||
|
|
||||||
def skip_unless_zmq(func):
|
|
||||||
""" Decorator that skips a test if we're using the pyevent hub."""
|
|
||||||
return skip_unless(using_zmq)(func)
|
|
||||||
|
|
||||||
|
|
||||||
class TestUpstreamDownStream(LimitedTestCase):
|
class TestUpstreamDownStream(LimitedTestCase):
|
||||||
|
|
||||||
|
Reference in New Issue
Block a user