File Manager

003 File Manager

Current Path: /usr/local/lib/python3.8/site-packages/zmq/tests

usr / local / lib / python3.8 / site-packages / zmq / tests /

📁 ..
📄 __init__.py(7.75 KB)
📁 __pycache__
📄 conftest.py(40 B)
📄 test_asyncio.py(15.57 KB)
📄 test_auth.py(20.72 KB)
📄 test_cffi_backend.py(8.79 KB)
📄 test_constants.py(4.49 KB)
📄 test_context.py(12.1 KB)
📄 test_cython.py(1.03 KB)
📄 test_decorators.py(9.4 KB)
📄 test_device.py(6.02 KB)
📄 test_draft.py(1.42 KB)
📄 test_error.py(1.17 KB)
📄 test_etc.py(545 B)
📄 test_future.py(10.62 KB)
📄 test_imports.py(1.76 KB)
📄 test_includes.py(997 B)
📄 test_ioloop.py(3.84 KB)
📄 test_log.py(6.51 KB)
📄 test_message.py(11.68 KB)
📄 test_monitor.py(3 KB)
📄 test_monqueue.py(7.97 KB)
📄 test_multipart.py(941 B)
📄 test_mypy.py(1.57 KB)
📄 test_pair.py(1.24 KB)
📄 test_poll.py(7.17 KB)
📄 test_proxy_steerable.py(3.7 KB)
📄 test_pubsub.py(1.07 KB)
📄 test_reqrep.py(1.8 KB)
📄 test_retry_eintr.py(2.88 KB)
📄 test_security.py(8 KB)
📄 test_socket.py(23.09 KB)
📄 test_ssh.py(235 B)
📄 test_version.py(1.3 KB)
📄 test_win32_shim.py(1.68 KB)
📄 test_z85.py(2.15 KB)
📄 test_zmqstream.py(2.35 KB)

Editing: test_draft.py

# -*- 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

003 File Manager

Editing: test_draft.py

Upload File

Create Folder