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_proxy_steerable.py (3922B)

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

import time
import struct

import zmq
from zmq import devices
from zmq.tests import BaseZMQTestCase, SkipTest, PYPY

if PYPY:
    # cleanup of shared Context doesn't work on PyPy
    devices.Device.context_factory = zmq.Context

class TestProxySteerable(BaseZMQTestCase):

def test_proxy_steerable(self):
        if zmq.zmq_version_info() < (4, 1):
            raise SkipTest("Steerable Proxies only in libzmq >= 4.1")
        dev = devices.ThreadProxySteerable(
            zmq.PULL,
            zmq.PUSH,
            zmq.PUSH,
            zmq.PAIR
        )
        iface = 'tcp://127.0.0.1'
        port = dev.bind_in_to_random_port(iface)
        port2 = dev.bind_out_to_random_port(iface)
        port3 = dev.bind_mon_to_random_port(iface)
        port4 = dev.bind_ctrl_to_random_port(iface)
        dev.start()
        time.sleep(0.25)
        msg = b'hello'
        push = self.context.socket(zmq.PUSH)
        push.connect("%s:%i" % (iface, port))
        pull = self.context.socket(zmq.PULL)
        pull.connect("%s:%i" % (iface, port2))
        mon = self.context.socket(zmq.PULL)
        mon.connect("%s:%i" % (iface, port3))
        ctrl = self.context.socket(zmq.PAIR)
        ctrl.connect("%s:%i" % (iface, port4))
        push.send(msg)
        self.sockets.extend([push, pull, mon, ctrl])
        self.assertEqual(msg, self.recv(pull))
        self.assertEqual(msg, self.recv(mon))
        ctrl.send(b'TERMINATE')
        dev.join()

def test_proxy_steerable_bind_to_random_with_args(self):
        if zmq.zmq_version_info() < (4, 1):
            raise SkipTest("Steerable Proxies only in libzmq >= 4.1")
        dev = devices.ThreadProxySteerable(
            zmq.PULL,
            zmq.PUSH,
            zmq.PUSH,
            zmq.PAIR
        )
        iface = 'tcp://127.0.0.1'
        ports = []
        min, max = 5000, 5050
        ports.extend([
            dev.bind_in_to_random_port(iface, min_port=min, max_port=max),
            dev.bind_out_to_random_port(iface, min_port=min, max_port=max),
            dev.bind_mon_to_random_port(iface, min_port=min, max_port=max),
            dev.bind_ctrl_to_random_port(iface, min_port=min, max_port=max)
        ])
        for port in ports:
            if port < min or port > max:
                self.fail('Unexpected port number: %i' % port)

def test_proxy_steerable_statistics(self):
        if zmq.zmq_version_info() < (4, 3):
            raise SkipTest("STATISTICS only in libzmq >= 4.3")
        dev = devices.ThreadProxySteerable(
            zmq.PULL,
            zmq.PUSH,
            zmq.PUSH,
            zmq.PAIR
        )
        iface = 'tcp://127.0.0.1'
        port = dev.bind_in_to_random_port(iface)
        port2 = dev.bind_out_to_random_port(iface)
        port3 = dev.bind_mon_to_random_port(iface)
        port4 = dev.bind_ctrl_to_random_port(iface)
        dev.start()
        time.sleep(0.25)
        msg = b'hello'
        push = self.context.socket(zmq.PUSH)
        push.connect("%s:%i" % (iface, port))
        pull = self.context.socket(zmq.PULL)
        pull.connect("%s:%i" % (iface, port2))
        mon = self.context.socket(zmq.PULL)
        mon.connect("%s:%i" % (iface, port3))
        ctrl = self.context.socket(zmq.PAIR)
        ctrl.connect("%s:%i" % (iface, port4))
        push.send(msg)
        self.sockets.extend([push, pull, mon, ctrl])
        self.assertEqual(msg, self.recv(pull))
        self.assertEqual(msg, self.recv(mon))
        ctrl.send(b'STATISTICS')
        stats = self.recv_multipart(ctrl)
        stats_int = [struct.unpack("=Q", x)[0] for x in stats]
        self.assertEqual(1, stats_int[0])
        self.assertEqual(len(msg), stats_int[1])
        self.assertEqual(1, stats_int[6])
        self.assertEqual(len(msg), stats_int[7])
        ctrl.send(b'TERMINATE')
        dev.join()

# Copyright (C) PyZMQ Developers # Distributed under the terms of the Modified BSD License. import time import struct import zmq from zmq import devices from zmq.tests import BaseZMQTestCase, SkipTest, PYPY if PYPY: # cleanup of shared Context doesn't work on PyPy devices.Device.context_factory = zmq.Context class TestProxySteerable(BaseZMQTestCase): def test_proxy_steerable(self): if zmq.zmq_version_info() < (4, 1): raise SkipTest("Steerable Proxies only in libzmq >= 4.1") dev = devices.ThreadProxySteerable( zmq.PULL, zmq.PUSH, zmq.PUSH, zmq.PAIR ) iface = 'tcp://127.0.0.1' port = dev.bind_in_to_random_port(iface) port2 = dev.bind_out_to_random_port(iface) port3 = dev.bind_mon_to_random_port(iface) port4 = dev.bind_ctrl_to_random_port(iface) dev.start() time.sleep(0.25) msg = b'hello' push = self.context.socket(zmq.PUSH) push.connect("%s:%i" % (iface, port)) pull = self.context.socket(zmq.PULL) pull.connect("%s:%i" % (iface, port2)) mon = self.context.socket(zmq.PULL) mon.connect("%s:%i" % (iface, port3)) ctrl = self.context.socket(zmq.PAIR) ctrl.connect("%s:%i" % (iface, port4)) push.send(msg) self.sockets.extend([push, pull, mon, ctrl]) self.assertEqual(msg, self.recv(pull)) self.assertEqual(msg, self.recv(mon)) ctrl.send(b'TERMINATE') dev.join() def test_proxy_steerable_bind_to_random_with_args(self): if zmq.zmq_version_info() < (4, 1): raise SkipTest("Steerable Proxies only in libzmq >= 4.1") dev = devices.ThreadProxySteerable( zmq.PULL, zmq.PUSH, zmq.PUSH, zmq.PAIR ) iface = 'tcp://127.0.0.1' ports = [] min, max = 5000, 5050 ports.extend([ dev.bind_in_to_random_port(iface, min_port=min, max_port=max), dev.bind_out_to_random_port(iface, min_port=min, max_port=max), dev.bind_mon_to_random_port(iface, min_port=min, max_port=max), dev.bind_ctrl_to_random_port(iface, min_port=min, max_port=max) ]) for port in ports: if port < min or port > max: self.fail('Unexpected port number: %i' % port) def test_proxy_steerable_statistics(self): if zmq.zmq_version_info() < (4, 3): raise SkipTest("STATISTICS only in libzmq >= 4.3") dev = devices.ThreadProxySteerable( zmq.PULL, zmq.PUSH, zmq.PUSH, zmq.PAIR ) iface = 'tcp://127.0.0.1' port = dev.bind_in_to_random_port(iface) port2 = dev.bind_out_to_random_port(iface) port3 = dev.bind_mon_to_random_port(iface) port4 = dev.bind_ctrl_to_random_port(iface) dev.start() time.sleep(0.25) msg = b'hello' push = self.context.socket(zmq.PUSH) push.connect("%s:%i" % (iface, port)) pull = self.context.socket(zmq.PULL) pull.connect("%s:%i" % (iface, port2)) mon = self.context.socket(zmq.PULL) mon.connect("%s:%i" % (iface, port3)) ctrl = self.context.socket(zmq.PAIR) ctrl.connect("%s:%i" % (iface, port4)) push.send(msg) self.sockets.extend([push, pull, mon, ctrl]) self.assertEqual(msg, self.recv(pull)) self.assertEqual(msg, self.recv(mon)) ctrl.send(b'STATISTICS') stats = self.recv_multipart(ctrl) stats_int = [struct.unpack("=Q", x)[0] for x in stats] self.assertEqual(1, stats_int[0]) self.assertEqual(len(msg), stats_int[1]) self.assertEqual(1, stats_int[6]) self.assertEqual(len(msg), stats_int[7]) ctrl.send(b'TERMINATE') dev.join()