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_retry_eintr.py (2960B)

# -*- 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 = .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
    
    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")

# -*- 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 = .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 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")