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_zmqstream.py (2439B)

# -*- coding: utf8 -*-
# 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

from unittest import TestCase

import pytest
import zmq
try:
    import tornado
    from tornado import gen
    from zmq.eventloop import ioloop, zmqstream
except ImportError:
    tornado = None

class TestZMQStream(TestCase):

def setUp(self):
        if tornado is None:
            pytest.skip()
        if asyncio:
            asyncio.set_event_loop(asyncio.new_event_loop())
        self.context = zmq.Context()
        self.loop = ioloop.IOLoop()
        self.loop.make_current()
        self.push = zmqstream.ZMQStream(self.context.socket(zmq.PUSH))
        self.pull = zmqstream.ZMQStream(self.context.socket(zmq.PULL))
        port = self.push.bind_to_random_port('tcp://127.0.0.1')
        self.pull.connect('tcp://127.0.0.1:%i' % port)
        self.stream = self.push

def tearDown(self):
        self.loop.close(all_fds=True)
        self.context.term()
        ioloop.IOLoop.clear_current()

def run_until_timeout(self, timeout=10):
        timed_out = []
        @gen.coroutine
        def sleep_timeout():
            yield gen.sleep(timeout)
            timed_out[:] = ['timed out']
            self.loop.stop()
        self.loop.add_callback(lambda : sleep_timeout())
        self.loop.start()
        assert not timed_out

def test_callable_check(self):
        """Ensure callable check works (py3k)."""
        
        self.stream.on_send(lambda *args: None)
        self.stream.on_recv(lambda *args: None)
        self.assertRaises(AssertionError, self.stream.on_recv, 1)
        self.assertRaises(AssertionError, self.stream.on_send, 1)
        self.assertRaises(AssertionError, self.stream.on_recv, zmq)

def test_on_recv_basic(self):
        sent = [b'basic']
        def callback(msg):
            assert msg == sent
            self.loop.stop()
        self.loop.add_callback(lambda : self.push.send_multipart(sent))
        self.pull.on_recv(callback)
        self.run_until_timeout()

def test_on_recv_wake(self):
        sent = [b'wake']
        def callback(msg):
            assert msg == sent
            self.loop.stop()
        self.pull.on_recv(callback)
        self.loop.call_later(1, lambda : self.push.send_multipart(sent))
        self.run_until_timeout()

# -*- coding: utf8 -*- # 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 from unittest import TestCase import pytest import zmq try: import tornado from tornado import gen from zmq.eventloop import ioloop, zmqstream except ImportError: tornado = None class TestZMQStream(TestCase): def setUp(self): if tornado is None: pytest.skip() if asyncio: asyncio.set_event_loop(asyncio.new_event_loop()) self.context = zmq.Context() self.loop = ioloop.IOLoop() self.loop.make_current() self.push = zmqstream.ZMQStream(self.context.socket(zmq.PUSH)) self.pull = zmqstream.ZMQStream(self.context.socket(zmq.PULL)) port = self.push.bind_to_random_port('tcp://127.0.0.1') self.pull.connect('tcp://127.0.0.1:%i' % port) self.stream = self.push def tearDown(self): self.loop.close(all_fds=True) self.context.term() ioloop.IOLoop.clear_current() def run_until_timeout(self, timeout=10): timed_out = [] @gen.coroutine def sleep_timeout(): yield gen.sleep(timeout) timed_out[:] = ['timed out'] self.loop.stop() self.loop.add_callback(lambda : sleep_timeout()) self.loop.start() assert not timed_out def test_callable_check(self): """Ensure callable check works (py3k).""" self.stream.on_send(lambda *args: None) self.stream.on_recv(lambda *args: None) self.assertRaises(AssertionError, self.stream.on_recv, 1) self.assertRaises(AssertionError, self.stream.on_send, 1) self.assertRaises(AssertionError, self.stream.on_recv, zmq) def test_on_recv_basic(self): sent = [b'basic'] def callback(msg): assert msg == sent self.loop.stop() self.loop.add_callback(lambda : self.push.send_multipart(sent)) self.pull.on_recv(callback) self.run_until_timeout() def test_on_recv_wake(self): sent = [b'wake'] def callback(msg): assert msg == sent self.loop.stop() self.pull.on_recv(callback) self.loop.call_later(1, lambda : self.push.send_multipart(sent)) self.run_until_timeout()