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_win32_shim.py (1741B)

from __future__ import print_function

import os
import time
import sys
from functools import wraps

from pytest import mark

from zmq.tests import BaseZMQTestCase
from zmq.utils.win32 import allow_interrupt

def count_calls(f):
    @wraps(f)
    def _(*args, **kwds):
        try:
            return f(*args, **kwds)
        finally:
            _.__calls__ += 1
    _.__calls__ = 0
    return _

@mark.new_console
class TestWindowsConsoleControlHandler(BaseZMQTestCase):

@mark.new_console
    @mark.skipif(
        not sys.platform.startswith('win'),
        reason='Windows only test')
    def test_handler(self):
        @count_calls
        def interrupt_polling():
            print('Caught CTRL-C!')

from ctypes import windll
        from ctypes.wintypes import BOOL, DWORD

kernel32 = windll.LoadLibrary('kernel32')

# <http://msdn.microsoft.com/en-us/library/ms683155.aspx>
        GenerateConsoleCtrlEvent = kernel32.GenerateConsoleCtrlEvent
        GenerateConsoleCtrlEvent.argtypes = (DWORD, DWORD)
        GenerateConsoleCtrlEvent.restype = BOOL

# Simulate CTRL-C event while handler is active.
        try:
            with allow_interrupt(interrupt_polling) as context:
                result = GenerateConsoleCtrlEvent(0, 0)
                # Sleep so that we give time to the handler to
                # capture the Ctrl-C event.
                time.sleep(0.5)
        except KeyboardInterrupt:
            pass
        else:
            if result == 0:
                raise WindowsError()
            else:
                self.fail('Expecting `KeyboardInterrupt` exception!')

# Make sure our handler was called.
        self.assertEqual(interrupt_polling.__calls__, 1)

from __future__ import print_function import os import time import sys from functools import wraps from pytest import mark from zmq.tests import BaseZMQTestCase from zmq.utils.win32 import allow_interrupt def count_calls(f): @wraps(f) def _(*args, **kwds): try: return f(*args, **kwds) finally: _.__calls__ += 1 _.__calls__ = 0 return _ @mark.new_console class TestWindowsConsoleControlHandler(BaseZMQTestCase): @mark.new_console @mark.skipif( not sys.platform.startswith('win'), reason='Windows only test') def test_handler(self): @count_calls def interrupt_polling(): print('Caught CTRL-C!') from ctypes import windll from ctypes.wintypes import BOOL, DWORD kernel32 = windll.LoadLibrary('kernel32') # GenerateConsoleCtrlEvent = kernel32.GenerateConsoleCtrlEvent GenerateConsoleCtrlEvent.argtypes = (DWORD, DWORD) GenerateConsoleCtrlEvent.restype = BOOL # Simulate CTRL-C event while handler is active. try: with allow_interrupt(interrupt_polling) as context: result = GenerateConsoleCtrlEvent(0, 0) # Sleep so that we give time to the handler to # capture the Ctrl-C event. time.sleep(0.5) except KeyboardInterrupt: pass else: if result == 0: raise WindowsError() else: self.fail('Expecting `KeyboardInterrupt` exception!') # Make sure our handler was called. self.assertEqual(interrupt_polling.__calls__, 1)