FM / tests

/opt/imh-python/lib/python3.9/site-packages/zmq/tests/

/opt/imh-python/lib/python3.9/site-packages/zmq/tests

Name	Size	Mode	Actions
asyncio/	-	0755	rm
__pycache__/	-	0755	rm
conftest.py	366	0644	edit dl rm
test_auth.py	20712	0644	edit dl rm
test_cffi_backend.py	9481	0644	edit dl rm
test_constants.py	4570	0644	edit dl rm
test_context.py	12027	0644	edit dl rm
test_cython.py	1048	0644	edit dl rm
test_decorators.py	9495	0644	edit dl rm
test_device.py	6169	0644	edit dl rm
test_draft.py	1471	0644	edit dl rm
test_error.py	1260	0644	edit dl rm
test_etc.py	525	0644	edit dl rm
test_future.py	11159	0644	edit dl rm
test_imports.py	1805	0644	edit dl rm
test_includes.py	1013	0644	edit dl rm
test_ioloop.py	3965	0644	edit dl rm
test_log.py	6697	0644	edit dl rm
test_message.py	11091	0644	edit dl rm
test_monitor.py	3044	0644	edit dl rm
test_monqueue.py	8223	0644	edit dl rm
test_multipart.py	944	0644	edit dl rm
test_pair.py	1260	0644	edit dl rm
test_poll.py	7260	0644	edit dl rm
test_proxy_steerable.py	3922	0644	edit dl rm
test_pubsub.py	1089	0644	edit dl rm
test_reqrep.py	1841	0644	edit dl rm
test_retry_eintr.py	2960	0644	edit dl rm
test_security.py	8141	0644	edit dl rm
test_socket.py	21653	0644	edit dl rm
test_ssh.py	234	0644	edit dl rm
test_version.py	1334	0644	edit dl rm
test_win32_shim.py	1741	0644	edit dl rm
test_z85.py	2232	0644	edit dl rm
test_zmqstream.py	2439	0644	edit dl rm
__init__.py	6382	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/zmq/tests/test_ioloop.py (3965B)

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

from __future__ import absolute_import
try:
    import asyncio
except ImportError:
    asyncio = None
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
    _tornado = True
except ImportError:
    _tornado = False

# 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

cmd: