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_retry_eintr.py

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

import signal
import time
from threading import Thread

from pytest import mark

import zmq
from zmq.tests import BaseZMQTestCase, SkipTest, skip_pypy
from zmq.utils.strtypes import b

# Partially based on EINTRBaseTest from CPython 3.5 eintr_tester

class TestEINTRSysCall(BaseZMQTestCase):
    """Base class for EINTR tests."""

# delay for initial signal delivery
    signal_delay = 0.1
    # timeout for tests. Must be > signal_delay
    timeout = 0.25
    timeout_ms = int(timeout * 1e3)

def alarm(self, t=None):
        """start a timer to fire only once

like signal.alarm, but with better resolution than integer seconds.
        """
        if not hasattr(signal, 'setitimer'):
            raise SkipTest('EINTR tests require setitimer')
        if t is None:
            t = self.signal_delay
        self.timer_fired = False
        self.orig_handler = signal.signal(signal.SIGALRM, self.stop_timer)
        # signal_period ignored, since only one timer event is allowed to fire
        signal.setitimer(signal.ITIMER_REAL, t, 1000)

def stop_timer(self, *args):
        self.timer_fired = True
        signal.setitimer(signal.ITIMER_REAL, 0, 0)
        signal.signal(signal.SIGALRM, self.orig_handler)

@mark.skipif(not hasattr(zmq, 'RCVTIMEO'), reason="requires RCVTIMEO")
    def test_retry_recv(self):
        pull = self.socket(zmq.PULL)
        pull.rcvtimeo = self.timeout_ms
        self.alarm()
        self.assertRaises(zmq.Again, pull.recv)
        assert self.timer_fired

@mark.skipif(not hasattr(zmq, 'SNDTIMEO'), reason="requires SNDTIMEO")
    def test_retry_send(self):
        push = self.socket(zmq.PUSH)
        push.sndtimeo = self.timeout_ms
        self.alarm()
        self.assertRaises(zmq.Again, push.send, b('buf'))
        assert self.timer_fired

@mark.flaky(reruns=3)
    def test_retry_poll(self):
        x, y = self.create_bound_pair()
        poller = zmq.Poller()
        poller.register(x, zmq.POLLIN)
        self.alarm()

def send():
            time.sleep(2 * self.signal_delay)
            y.send(b('ping'))

t = Thread(target=send)
        t.start()
        evts = dict(poller.poll(2 * self.timeout_ms))
        t.join()
        assert x in evts
        assert self.timer_fired
        x.recv()

def test_retry_term(self):
        push = self.socket(zmq.PUSH)
        push.linger = self.timeout_ms
        push.connect('tcp://127.0.0.1:5555')
        push.send(b('ping'))
        time.sleep(0.1)
        self.alarm()
        self.context.destroy()
        assert self.timer_fired
        assert self.context.closed

def test_retry_getsockopt(self):
        raise SkipTest("TODO: find a way to interrupt getsockopt")

def test_retry_setsockopt(self):
        raise SkipTest("TODO: find a way to interrupt setsockopt")

003 File Manager

Editing: test_retry_eintr.py

Upload File

Create Folder