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_draft.py (1471B)

# -*- coding: utf8 -*-
# Copyright (C) PyZMQ Developers
# Distributed under the terms of the Modified BSD License.

import os
import platform
import time

import pytest
import zmq
from zmq.tests import (
    BaseZMQTestCase, skip_pypy
)

class TestDraftSockets(BaseZMQTestCase):
    def setUp(self):
        if not zmq.DRAFT_API:
            raise pytest.skip("draft api unavailable")
        super(TestDraftSockets, self).setUp()

def test_client_server(self):
        client, server = self.create_bound_pair(zmq.CLIENT, zmq.SERVER)
        client.send(b'request')
        msg = self.recv(server, copy=False)
        assert msg.routing_id is not None
        server.send(b'reply', routing_id=msg.routing_id)
        reply = self.recv(client)
        assert reply == b'reply'

def test_radio_dish(self):
        dish, radio = self.create_bound_pair(zmq.DISH, zmq.RADIO)
        dish.rcvtimeo = 250
        group = 'mygroup'
        dish.join(group)
        received_count = 0
        received = set()
        sent = set()
        for i in range(10):
            msg = str(i).encode('ascii')
            sent.add(msg)
            radio.send(msg, group=group)
            try:
                recvd = dish.recv()
            except zmq.Again:
                time.sleep(0.1)
            else:
                received.add(recvd)
                received_count += 1
        # assert that we got *something*
        assert len(received.intersection(sent)) >= 5

cmd: