@@ -0,0 +1,3094 @@
+#-----------------------------------------------------------------------------%0A#  Copyright (c) 2010-2012 Brian Granger, Min Ragan-Kelley%0A#%0A#  This file is part of pyzmq%0A#%0A#  Distributed under the terms of the New BSD License.  The full license is in%0A#  the file COPYING.BSD, distributed as part of this software.%0A#-----------------------------------------------------------------------------%0A%0A#-----------------------------------------------------------------------------%0A# Imports%0A#-----------------------------------------------------------------------------%0A%0Aimport time%0Aimport os%0Aimport threading%0A%0Aimport zmqpy as zmq%0Afrom .__init__ import BaseZMQTestCase%0Afrom zmqpy.eventloop import ioloop%0A%0A%0A#-----------------------------------------------------------------------------%0A# Tests%0A#-----------------------------------------------------------------------------%0Adef printer():%0A    os.system(%22say hello%22)%0A    raise Exception%0A    print (time.time())%0A%0Aclass Delay(threading.Thread):%0A    def __init__(self, f, delay=1):%0A        self.f=f%0A        self.delay=delay%0A        self.aborted=False%0A        self.cond=threading.Condition()%0A        super(Delay, self).__init__()%0A%0A    def run(self):%0A        self.cond.acquire()%0A        self.cond.wait(self.delay)%0A        self.cond.release()%0A        if not self.aborted:%0A            self.f()%0A%0A    def abort(self):%0A        self.aborted=True%0A        self.cond.acquire()%0A        self.cond.notify()%0A        self.cond.release()%0A%0Aclass TestIOLoop(BaseZMQTestCase):%0A%0A    def test_simple(self):%0A        %22%22%22simple IOLoop creation test%22%22%22%0A        loop = ioloop.IOLoop()%0A        dc = ioloop.DelayedCallback(loop.stop, 200, loop)%0A        pc = ioloop.DelayedCallback(lambda : None, 10, loop)%0A        pc.start()%0A        dc.start()%0A        t = Delay(loop.stop,1)%0A        t.start()%0A        loop.start()%0A        if t.isAlive():%0A            t.abort()%0A        else:%0A            self.fail(%22IOLoop failed to exit%22)%0A%0A    def test_timeout_compare(self):%0A        %22%22%22test timeout comparisons%22%22%22%0A        t = ioloop._Timeout(1,2)%0A        t2 = ioloop._Timeout(1,3)%0A        self.assertEquals(t %3C t2, id(t) %3C id(t2))%0A        t2 = ioloop._Timeout(2,1)%0A        self.assertTrue(t %3C t2)%0A%0A    def test_poller_events(self):%0A        %22%22%22Tornado poller implementation maps events correctly%22%22%22%0A        req,rep = self.create_bound_pair(zmq.REQ, zmq.REP)%0A        poller = ioloop.ZMQPoller()%0A        poller.register(req, ioloop.IOLoop.READ)%0A        poller.register(rep, ioloop.IOLoop.READ)%0A        events = dict(poller.poll(0))%0A        self.assertEquals(events.get(rep), None)%0A        self.assertEquals(events.get(req), None)%0A%0A        poller.register(req, ioloop.IOLoop.WRITE)%0A        poller.register(rep, ioloop.IOLoop.WRITE)%0A        events = dict(poller.poll(1))%0A        self.assertEquals(events.get(req), ioloop.IOLoop.WRITE)%0A        self.assertEquals(events.get(rep), None)%0A%0A        poller.register(rep, ioloop.IOLoop.READ)%0A        req.send(b'hi')%0A        events = dict(poller.poll(1))%0A        self.assertEquals(events.get(rep), ioloop.IOLoop.READ)%0A        self.assertEquals(events.get(req), None)%0A%0A

add pyzmq IOLoop tests