/usr/lib/python2.7/dist-packages/zmq/tests/test_ioloop.py is in python-zmq 16.0.2-2build2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 | # Copyright (C) PyZMQ Developers
# Distributed under the terms of the Modified BSD License.
import time
import os
import threading
import pytest
import zmq
from zmq.tests import BaseZMQTestCase, have_gevent
from zmq.eventloop import ioloop
try:
from tornado.ioloop import IOLoop as BaseIOLoop
except ImportError:
from zmq.eventloop.minitornado.ioloop import IOLoop as BaseIOLoop
def printer():
os.system("say hello")
raise Exception
print (time.time())
class Delay(threading.Thread):
def __init__(self, f, delay=1):
self.f=f
self.delay=delay
self.aborted=False
self.cond=threading.Condition()
super(Delay, self).__init__()
def run(self):
self.cond.acquire()
self.cond.wait(self.delay)
self.cond.release()
if not self.aborted:
self.f()
def abort(self):
self.aborted=True
self.cond.acquire()
self.cond.notify()
self.cond.release()
class TestIOLoop(BaseZMQTestCase):
def tearDown(self):
super(TestIOLoop, self).tearDown()
BaseIOLoop.clear_current()
BaseIOLoop.clear_instance()
def test_simple(self):
"""simple IOLoop creation test"""
loop = ioloop.IOLoop()
dc = ioloop.PeriodicCallback(loop.stop, 200, loop)
pc = ioloop.PeriodicCallback(lambda : None, 10, loop)
pc.start()
dc.start()
t = Delay(loop.stop,1)
t.start()
loop.start()
if t.isAlive():
t.abort()
else:
self.fail("IOLoop failed to exit")
def test_poller_events(self):
"""Tornado poller implementation maps events correctly"""
req,rep = self.create_bound_pair(zmq.REQ, zmq.REP)
poller = ioloop.ZMQPoller()
poller.register(req, ioloop.IOLoop.READ)
poller.register(rep, ioloop.IOLoop.READ)
events = dict(poller.poll(0))
self.assertEqual(events.get(rep), None)
self.assertEqual(events.get(req), None)
poller.register(req, ioloop.IOLoop.WRITE)
poller.register(rep, ioloop.IOLoop.WRITE)
events = dict(poller.poll(1))
self.assertEqual(events.get(req), ioloop.IOLoop.WRITE)
self.assertEqual(events.get(rep), None)
poller.register(rep, ioloop.IOLoop.READ)
req.send(b'hi')
events = dict(poller.poll(1))
self.assertEqual(events.get(rep), ioloop.IOLoop.READ)
self.assertEqual(events.get(req), None)
def test_instance(self):
"""Green IOLoop.instance returns the right object"""
loop = ioloop.IOLoop.instance()
assert isinstance(loop, ioloop.IOLoop)
base_loop = BaseIOLoop.instance()
assert base_loop is loop
def test_current(self):
"""Green IOLoop.current returns the right object"""
loop = ioloop.IOLoop.current()
assert isinstance(loop, ioloop.IOLoop)
base_loop = BaseIOLoop.current()
assert base_loop is loop
def test_close_all(self):
"""Test close(all_fds=True)"""
loop = ioloop.IOLoop.instance()
req,rep = self.create_bound_pair(zmq.REQ, zmq.REP)
loop.add_handler(req, lambda msg: msg, ioloop.IOLoop.READ)
loop.add_handler(rep, lambda msg: msg, ioloop.IOLoop.READ)
self.assertEqual(req.closed, False)
self.assertEqual(rep.closed, False)
loop.close(all_fds=True)
self.assertEqual(req.closed, True)
self.assertEqual(rep.closed, True)
if have_gevent:
import zmq.green.eventloop.ioloop as green_ioloop
class TestIOLoopGreen(BaseZMQTestCase):
def test_instance(self):
"""Green IOLoop.instance returns the right object"""
loop = green_ioloop.IOLoop.instance()
assert isinstance(loop, green_ioloop.IOLoop)
base_loop = BaseIOLoop.instance()
assert base_loop is loop
def test_current(self):
"""Green IOLoop.current returns the right object"""
loop = green_ioloop.IOLoop.current()
assert isinstance(loop, green_ioloop.IOLoop)
base_loop = BaseIOLoop.current()
assert base_loop is loop
|