284 lines
8.7 KiB
Python
284 lines
8.7 KiB
Python
from eventlet import event, spawn, sleep, patcher
|
|
from eventlet.hubs import get_hub, _threadlocal, use_hub
|
|
from nose.tools import *
|
|
from tests import mock, LimitedTestCase, using_pyevent, skip_unless
|
|
from unittest import TestCase
|
|
|
|
from threading import Thread
|
|
try:
|
|
from eventlet.green import zmq
|
|
except ImportError:
|
|
zmq = {} # for systems lacking zmq, skips tests instead of barfing
|
|
|
|
def zmq_supported(_):
|
|
try:
|
|
import zmq
|
|
except ImportError:
|
|
return False
|
|
return not using_pyevent(_)
|
|
|
|
class TestUpstreamDownStream(LimitedTestCase):
|
|
|
|
sockets = []
|
|
|
|
def tearDown(self):
|
|
self.clear_up_sockets()
|
|
super(TestUpstreamDownStream, self).tearDown()
|
|
|
|
def create_bound_pair(self, type1, type2, interface='tcp://127.0.0.1'):
|
|
"""Create a bound socket pair using a random port."""
|
|
self.context = context = zmq.Context()
|
|
s1 = context.socket(type1)
|
|
port = s1.bind_to_random_port(interface)
|
|
s2 = context.socket(type2)
|
|
s2.connect('%s:%s' % (interface, port))
|
|
self.sockets = [s1, s2]
|
|
return s1, s2, port
|
|
|
|
def clear_up_sockets(self):
|
|
for sock in self.sockets:
|
|
sock.close()
|
|
|
|
def assertRaisesErrno(self, errno, func, *args):
|
|
try:
|
|
func(*args)
|
|
except zmq.ZMQError, e:
|
|
self.assertEqual(e.errno, errno, "wrong error raised, expected '%s' \
|
|
got '%s'" % (zmq.ZMQError(errno), zmq.ZMQError(e.errno)))
|
|
else:
|
|
self.fail("Function did not raise any error")
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_recv_spawned_before_send_is_non_blocking(self):
|
|
req, rep, port = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
|
|
# req.connect(ipc)
|
|
# rep.bind(ipc)
|
|
sleep()
|
|
msg = dict(res=None)
|
|
done = event.Event()
|
|
|
|
def rx():
|
|
msg['res'] = rep.recv()
|
|
done.send('done')
|
|
|
|
spawn(rx)
|
|
req.send('test')
|
|
done.wait()
|
|
self.assertEqual(msg['res'], 'test')
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_close_socket_raises_enotsup(self):
|
|
req, rep, port = self.create_bound_pair(zmq.PAIR, zmq.PAIR)
|
|
|
|
rep.close()
|
|
req.close()
|
|
self.assertRaisesErrno(zmq.ENOTSUP, rep.recv)
|
|
self.assertRaisesErrno(zmq.ENOTSUP, req.send, 'test')
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_send_1k_req_rep(self):
|
|
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
|
|
sleep()
|
|
done = event.Event()
|
|
|
|
def tx():
|
|
tx_i = 0
|
|
req.send(str(tx_i))
|
|
while req.recv() != 'done':
|
|
tx_i += 1
|
|
req.send(str(tx_i))
|
|
|
|
def rx():
|
|
while True:
|
|
rx_i = rep.recv()
|
|
if rx_i == "1000":
|
|
rep.send('done')
|
|
sleep()
|
|
done.send(0)
|
|
break
|
|
rep.send('i')
|
|
spawn(tx)
|
|
spawn(rx)
|
|
final_i = done.wait()
|
|
self.assertEqual(final_i, 0)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_send_1k_push_pull(self):
|
|
down, up, port = self.create_bound_pair(zmq.PUSH, zmq.PULL)
|
|
sleep()
|
|
|
|
done = event.Event()
|
|
|
|
def tx():
|
|
tx_i = 0
|
|
while tx_i <= 1000:
|
|
tx_i += 1
|
|
down.send(str(tx_i))
|
|
|
|
def rx():
|
|
while True:
|
|
rx_i = up.recv()
|
|
if rx_i == "1000":
|
|
done.send(0)
|
|
break
|
|
spawn(tx)
|
|
spawn(rx)
|
|
final_i = done.wait()
|
|
self.assertEqual(final_i, 0)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_send_1k_pub_sub(self):
|
|
pub, sub_all, port = self.create_bound_pair(zmq.PUB, zmq.SUB)
|
|
sub1 = self.context.socket(zmq.SUB)
|
|
sub2 = self.context.socket(zmq.SUB)
|
|
self.sockets.extend([sub1, sub2])
|
|
addr = 'tcp://127.0.0.1:%s' % port
|
|
sub1.connect(addr)
|
|
sub2.connect(addr)
|
|
sub_all.setsockopt(zmq.SUBSCRIBE, '')
|
|
sub1.setsockopt(zmq.SUBSCRIBE, 'sub1')
|
|
sub2.setsockopt(zmq.SUBSCRIBE, 'sub2')
|
|
|
|
sub_all_done = event.Event()
|
|
sub1_done = event.Event()
|
|
sub2_done = event.Event()
|
|
|
|
sleep(0.2)
|
|
|
|
def rx(sock, done_evt, msg_count=10000):
|
|
count = 0
|
|
while count < msg_count:
|
|
msg = sock.recv()
|
|
sleep()
|
|
if 'LAST' in msg:
|
|
break
|
|
count += 1
|
|
|
|
done_evt.send(count)
|
|
|
|
def tx(sock):
|
|
for i in range(1, 1001):
|
|
msg = "sub%s %s" % ([2,1][i % 2], i)
|
|
sock.send(msg)
|
|
sleep()
|
|
sock.send('sub1 LAST')
|
|
sock.send('sub2 LAST')
|
|
|
|
spawn(rx, sub_all, sub_all_done)
|
|
spawn(rx, sub1, sub1_done)
|
|
spawn(rx, sub2, sub2_done)
|
|
spawn(tx, pub)
|
|
sub1_count = sub1_done.wait()
|
|
sub2_count = sub2_done.wait()
|
|
sub_all_count = sub_all_done.wait()
|
|
self.assertEqual(sub1_count, 500)
|
|
self.assertEqual(sub2_count, 500)
|
|
self.assertEqual(sub_all_count, 1000)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_change_subscription(self):
|
|
pub, sub, port = self.create_bound_pair(zmq.PUB, zmq.SUB)
|
|
sub.setsockopt(zmq.SUBSCRIBE, 'test')
|
|
|
|
sleep(0.2)
|
|
sub_done = event.Event()
|
|
|
|
def rx(sock, done_evt):
|
|
count = 0
|
|
sub = 'test'
|
|
while True:
|
|
msg = sock.recv()
|
|
sleep()
|
|
if 'DONE' in msg:
|
|
break
|
|
if 'LAST' in msg and sub == 'test':
|
|
sock.setsockopt(zmq.UNSUBSCRIBE, 'test')
|
|
sock.setsockopt(zmq.SUBSCRIBE, 'done')
|
|
sub = 'done'
|
|
count += 1
|
|
done_evt.send(count)
|
|
|
|
def tx(sock):
|
|
for i in range(1, 101):
|
|
msg = "test %s" % i
|
|
if i != 50:
|
|
sock.send(msg)
|
|
else:
|
|
sock.send('test LAST')
|
|
sleep()
|
|
sock.send('done DONE')
|
|
|
|
spawn(rx, sub, sub_done)
|
|
spawn(tx, pub)
|
|
|
|
rx_count = sub_done.wait()
|
|
self.assertEqual(rx_count, 50)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_recv_multipart_bug68(self):
|
|
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
|
|
msg = ['']
|
|
req.send_multipart(msg)
|
|
recieved_msg = rep.recv_multipart()
|
|
self.assertEqual(recieved_msg, msg)
|
|
|
|
# Send a message back the other way
|
|
msg2 = [""]
|
|
rep.send_multipart(msg2, copy=False)
|
|
# When receiving a copy it's a zmq.core.message.Message you get back
|
|
recieved_msg = req.recv_multipart(copy=False)
|
|
# So it needs to be converted to a string
|
|
# I'm calling str(m) consciously here; Message has a .data attribute
|
|
# but it's private __str__ appears to be the way to go
|
|
self.assertEqual([str(m) for m in recieved_msg], msg2)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_recv_noblock_bug76(self):
|
|
req, rep, port = self.create_bound_pair(zmq.REQ, zmq.REP)
|
|
self.assertRaisesErrno(zmq.EAGAIN, rep.recv, zmq.NOBLOCK)
|
|
self.assertRaisesErrno(zmq.EAGAIN, rep.recv, zmq.NOBLOCK, True)
|
|
|
|
|
|
|
|
class TestThreadedContextAccess(TestCase):
|
|
"""zmq's Context must be unique within a hub
|
|
|
|
The zeromq API documentation states:
|
|
All zmq sockets passed to the zmq_poll() function must share the same zmq
|
|
context and must belong to the thread calling zmq_poll()
|
|
|
|
As zmq_poll is what's eventually being called then we need to ensure that
|
|
all sockets that are going to be passed to zmq_poll (via hub.do_poll) are
|
|
in the same context
|
|
"""
|
|
if zmq: # don't call decorators if zmq module unavailable
|
|
@skip_unless(zmq_supported)
|
|
def test_context_factory_function(self):
|
|
ctx = zmq.Context()
|
|
self.assertTrue(ctx is not None)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_threadlocal_context(self):
|
|
context = zmq.Context()
|
|
self.assertEqual(context, _threadlocal.context)
|
|
next_context = zmq.Context()
|
|
self.assertTrue(context is next_context)
|
|
|
|
@skip_unless(zmq_supported)
|
|
def test_different_context_in_different_thread(self):
|
|
context = zmq.Context()
|
|
test_result = []
|
|
def assert_different(ctx):
|
|
try:
|
|
this_thread_context = zmq.Context()
|
|
except:
|
|
test_result.append('fail')
|
|
raise
|
|
test_result.append(ctx is this_thread_context)
|
|
|
|
Thread(target=assert_different, args=(context,)).start()
|
|
while not test_result:
|
|
sleep(0.1)
|
|
self.assertFalse(test_result[0])
|
|
|