File Manager

003 File Manager

Current Path: /usr/local/lib/python3.8/site-packages/zmq/tests

usr / local / lib / python3.8 / site-packages / zmq / tests /

📁 ..
📄 __init__.py(7.75 KB)
📁 __pycache__
📄 conftest.py(40 B)
📄 test_asyncio.py(15.57 KB)
📄 test_auth.py(20.72 KB)
📄 test_cffi_backend.py(8.79 KB)
📄 test_constants.py(4.49 KB)
📄 test_context.py(12.1 KB)
📄 test_cython.py(1.03 KB)
📄 test_decorators.py(9.4 KB)
📄 test_device.py(6.02 KB)
📄 test_draft.py(1.42 KB)
📄 test_error.py(1.17 KB)
📄 test_etc.py(545 B)
📄 test_future.py(10.62 KB)
📄 test_imports.py(1.76 KB)
📄 test_includes.py(997 B)
📄 test_ioloop.py(3.84 KB)
📄 test_log.py(6.51 KB)
📄 test_message.py(11.68 KB)
📄 test_monitor.py(3 KB)
📄 test_monqueue.py(7.97 KB)
📄 test_multipart.py(941 B)
📄 test_mypy.py(1.57 KB)
📄 test_pair.py(1.24 KB)
📄 test_poll.py(7.17 KB)
📄 test_proxy_steerable.py(3.7 KB)
📄 test_pubsub.py(1.07 KB)
📄 test_reqrep.py(1.8 KB)
📄 test_retry_eintr.py(2.88 KB)
📄 test_security.py(8 KB)
📄 test_socket.py(23.09 KB)
📄 test_ssh.py(235 B)
📄 test_version.py(1.3 KB)
📄 test_win32_shim.py(1.68 KB)
📄 test_z85.py(2.15 KB)
📄 test_zmqstream.py(2.35 KB)

Editing: test_ioloop.py

# Copyright (C) PyZMQ Developers
# Distributed under the terms of the Modified BSD License.

from __future__ import absolute_import

import asyncio
import time
import os
import threading

import pytest

import zmq
from zmq.tests import BaseZMQTestCase, have_gevent

try:
    from tornado.ioloop import IOLoop as BaseIOLoop
    from zmq.eventloop import ioloop
except ImportError:
    _tornado = False
else:
    _tornado = True

# tornado 5 with asyncio disables custom IOLoop implementations
t5asyncio = False
if _tornado:
    import tornado

if tornado.version_info >= (5,) and asyncio:
        t5asyncio = True

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):
    if _tornado:
        IOLoop = ioloop.IOLoop

def setUp(self):
        if not _tornado:
            pytest.skip("tornado required")
        super(TestIOLoop, self).setUp()
        if asyncio:
            asyncio.set_event_loop(asyncio.new_event_loop())

def tearDown(self):
        super(TestIOLoop, self).tearDown()
        BaseIOLoop.clear_current()
        BaseIOLoop.clear_instance()

def test_simple(self):
        """simple IOLoop creation test"""
        loop = self.IOLoop()
        loop.make_current()
        dc = ioloop.PeriodicCallback(loop.stop, 200)
        pc = ioloop.PeriodicCallback(lambda: None, 10)
        pc.start()
        dc.start()
        t = Delay(loop.stop, 1)
        t.start()
        loop.start()
        if t.is_alive():
            t.abort()
        else:
            self.fail("IOLoop failed to exit")

def test_instance(self):
        """IOLoop.instance returns the right object"""
        loop = self.IOLoop.instance()
        if not t5asyncio:
            assert isinstance(loop, self.IOLoop)
        base_loop = BaseIOLoop.instance()
        assert base_loop is loop

def test_current(self):
        """IOLoop.current returns the right object"""
        loop = ioloop.IOLoop.current()
        if not t5asyncio:
            assert isinstance(loop, self.IOLoop)
        base_loop = BaseIOLoop.current()
        assert base_loop is loop

def test_close_all(self):
        """Test close(all_fds=True)"""
        loop = self.IOLoop.current()
        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 and _tornado:
    import zmq.green.eventloop.ioloop as green_ioloop

class TestIOLoopGreen(TestIOLoop):
        IOLoop = green_ioloop.IOLoop

def xtest_instance(self):
            """Green IOLoop.instance returns the right object"""
            loop = self.IOLoop.instance()
            if not t5asyncio:
                assert isinstance(loop, self.IOLoop)
            base_loop = BaseIOLoop.instance()
            assert base_loop is loop

def xtest_current(self):
            """Green IOLoop.current returns the right object"""
            loop = self.IOLoop.current()
            if not t5asyncio:
                assert isinstance(loop, self.IOLoop)
            base_loop = BaseIOLoop.current()
            assert base_loop is loop

003 File Manager

Editing: test_ioloop.py

Upload File

Create Folder