Took sleep statements out of send/recv functions. Renamed tests to zmq_tests

1 files changed, 153 insertions, 0 deletions

diff --git a/tests/zmq_test.py b/tests/zmq_test.py
new file mode 100644
index 0000000..e7fbbe3
--- /dev/null
+++ b/tests/zmq_test.py
@@ -0,0 +1,153 @@
+from eventlet import event, spawn, sleep, patcher
+from eventlet.hubs import use_hub, get_hub, _threadlocal
+from eventlet.hubs.hub import READ, WRITE
+from eventlet.green import zmq
+from nose.tools import *
+from tests import mock, LimitedTestCase, skip_unless
+from unittest import TestCase
+
+from threading import Thread
+
+def using_zmq(_f):
+    return '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):
+
+    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 = get_hub().get_context()
+        s1 = context.socket(type1)
+        port = s1.bind_to_random_port(interface)
+        s2 = context.socket(type2)
+        s2.connect('%s:%s' % (interface, port))
+        return s1, s2
+
+    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
+    def test_recv_spawned_before_send_is_non_blocking(self):
+        req, rep = 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
+    def test_close_socket_raises_enotsup(self):
+        req, rep = 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
+    def test_send_1k_req_rep(self):
+        req, rep = 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
+    def test_send_1k_up_down(self):
+        down, up = self.create_bound_pair(zmq.DOWNSTREAM, zmq.UPSTREAM)
+        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)
+
+
+
+
+        
+
+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 insure that
+    all sockets that are going to be passed to zmq_poll (via hub.do_poll) are
+    in the same context
+    """
+
+    @skip_unless_zmq
+    def test_threadlocal_context(self):
+        hub = get_hub()
+        context = hub.get_context()
+        self.assertEqual(context, _threadlocal.context)
+        next_context = hub.get_context()
+        self.assertTrue(context is next_context)
+
+    @skip_unless_zmq
+    def test_different_context_in_different_thread(self):
+        context = get_hub().get_context()
+        test_result = []
+        def assert_different(ctx):
+            assert not hasattr(_threadlocal, 'hub')
+            hub = get_hub()
+            try:
+                this_thread_context = hub.get_context()
+            except:
+                test_result.append('fail')
+            test_result.append(ctx is this_thread_context)
+        Thread(target=assert_different, args=(context,)).start()
+        while not len(test_result):
+            pass
+        self.assertFalse(test_result[0])
+
+
+
+
+