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_reqrep.py (1841B)

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

from unittest import TestCase

import zmq
from zmq.tests import BaseZMQTestCase, have_gevent, GreenTest

class TestReqRep(BaseZMQTestCase):

def test_basic(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)

msg1 = b'message 1'
        msg2 = self.ping_pong(s1, s2, msg1)
        self.assertEqual(msg1, msg2)

def test_multiple(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)

for i in range(10):
            msg1 = i*b' '
            msg2 = self.ping_pong(s1, s2, msg1)
            self.assertEqual(msg1, msg2)

def test_bad_send_recv(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
        
        if zmq.zmq_version() != '2.1.8':
            # this doesn't work on 2.1.8
            for copy in (True,False):
                self.assertRaisesErrno(zmq.EFSM, s1.recv, copy=copy)
                self.assertRaisesErrno(zmq.EFSM, s2.send, b'asdf', copy=copy)

# I have to have this or we die on an Abort trap.
        msg1 = b'asdf'
        msg2 = self.ping_pong(s1, s2, msg1)
        self.assertEqual(msg1, msg2)

def test_json(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
        o = dict(a=10,b=list(range(10)))
        o2 = self.ping_pong_json(s1, s2, o)

def test_pyobj(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
        o = dict(a=10,b=range(10))
        o2 = self.ping_pong_pyobj(s1, s2, o)

def test_large_msg(self):
        s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP)
        msg1 = 10000*b'X'

for i in range(10):
            msg2 = self.ping_pong(s1, s2, msg1)
            self.assertEqual(msg1, msg2)

if have_gevent:
    class TestReqRepGreen(GreenTest, TestReqRep):
        pass

# Copyright (C) PyZMQ Developers # Distributed under the terms of the Modified BSD License. from unittest import TestCase import zmq from zmq.tests import BaseZMQTestCase, have_gevent, GreenTest class TestReqRep(BaseZMQTestCase): def test_basic(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) msg1 = b'message 1' msg2 = self.ping_pong(s1, s2, msg1) self.assertEqual(msg1, msg2) def test_multiple(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) for i in range(10): msg1 = i*b' ' msg2 = self.ping_pong(s1, s2, msg1) self.assertEqual(msg1, msg2) def test_bad_send_recv(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) if zmq.zmq_version() != '2.1.8': # this doesn't work on 2.1.8 for copy in (True,False): self.assertRaisesErrno(zmq.EFSM, s1.recv, copy=copy) self.assertRaisesErrno(zmq.EFSM, s2.send, b'asdf', copy=copy) # I have to have this or we die on an Abort trap. msg1 = b'asdf' msg2 = self.ping_pong(s1, s2, msg1) self.assertEqual(msg1, msg2) def test_json(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) o = dict(a=10,b=list(range(10))) o2 = self.ping_pong_json(s1, s2, o) def test_pyobj(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) o = dict(a=10,b=range(10)) o2 = self.ping_pong_pyobj(s1, s2, o) def test_large_msg(self): s1, s2 = self.create_bound_pair(zmq.REQ, zmq.REP) msg1 = 10000*b'X' for i in range(10): msg2 = self.ping_pong(s1, s2, msg1) self.assertEqual(msg1, msg2) if have_gevent: class TestReqRepGreen(GreenTest, TestReqRep): pass