only start the loop once the event loop is running close must not call signal setwakeupfd check that it s possible to cancel connectpipe wait for unset event with 0 5s timeout result should be false at timeout wait for set event result should be true immediately asyncio issue 195 cancelling a done waithandlefuture must not crash wait for unset event with a cancelled future cancellederror should be raised immediately asyncio issue 195 cancelling a waithandlefuture twice must not crash regression test for https bugs python orgissue39010 previously restarting a proactor event loop in certain states would lead to spurious connectionreseterrors being logged start an operation in another thread so that the selfpipe is used this is theoretically timingdependent the task in the executor must complete before our startstop cycles but in practice it seems to work every time shut everything down cleanly this is an important part of the test in issue 39010 the error occurred during loop close so we want to close the loop during the test instead of leaving it for teardown first wait for f to complete to avoid a future s result was never retrieved error now shut down the loop itself self closeloop also shuts down the loop s default executor regression test for https github compythoncpythonissues98793 regression test for https github compythoncpythonissues100573 see https github compythoncpythonpull100959discussionr1068533658 only start the loop once the event loop is running close must not call signal set_wakeup_fd check that it s possible to cancel connect_pipe wait for unset event with 0 5s timeout result should be false at timeout wait for set event result should be true immediately asyncio issue 195 cancelling a done _waithandlefuture must not crash wait for unset event with a cancelled future cancellederror should be raised immediately asyncio issue 195 cancelling a _waithandlefuture twice must not crash regression test for https bugs python org issue39010 previously restarting a proactor event loop in certain states would lead to spurious connectionreseterrors being logged start an operation in another thread so that the self pipe is used this is theoretically timing dependent the task in the executor must complete before our start stop cycles but in practice it seems to work every time shut everything down cleanly this is an important part of the test in issue 39010 the error occurred during loop close so we want to close the loop during the test instead of leaving it for teardown first wait for f to complete to avoid a future s result was never retrieved error now shut down the loop itself self close_loop also shuts down the loop s default executor regression test for https github com python cpython issues 98793 regression test for https github com python cpython issues 100573 see https github com python cpython pull 100959 discussion_r1068533658
import os
import signal
import socket
import sys
import time
import threading
import unittest
from unittest import mock

if sys.platform != 'win32':
    raise unittest.SkipTest('Windows only')

import _overlapped
import _winapi

import asyncio
from asyncio import windows_events
from test.test_asyncio import utils as test_utils


def tearDownModule():
    asyncio.set_event_loop_policy(None)


class UpperProto(asyncio.Protocol):
    def __init__(self):
        self.buf = []

    def connection_made(self, trans):
        self.trans = trans

    def data_received(self, data):
        self.buf.append(data)
        if b'\n' in data:
            self.trans.write(b''.join(self.buf).upper())
            self.trans.close()


class ProactorLoopCtrlC(test_utils.TestCase):

    def test_ctrl_c(self):

        def SIGINT_after_delay():
            time.sleep(0.1)
            signal.raise_signal(signal.SIGINT)

        thread = threading.Thread(target=SIGINT_after_delay)
        loop = asyncio.new_event_loop()
        try:
            
            loop.call_soon(thread.start)
            loop.run_forever()
            self.fail("should not fall through 'run_forever'")
        except KeyboardInterrupt:
            pass
        finally:
            self.close_loop(loop)
        thread.join()


class ProactorMultithreading(test_utils.TestCase):
    def test_run_from_nonmain_thread(self):
        finished = False

        async def coro():
            await asyncio.sleep(0)

        def func():
            nonlocal finished
            loop = asyncio.new_event_loop()
            loop.run_until_complete(coro())
            
            loop.close()
            finished = True

        thread = threading.Thread(target=func)
        thread.start()
        thread.join()
        self.assertTrue(finished)


class ProactorTests(test_utils.TestCase):

    def setUp(self):
        super().setUp()
        self.loop = asyncio.ProactorEventLoop()
        self.set_event_loop(self.loop)

    def test_close(self):
        a, b = socket.socketpair()
        trans = self.loop._make_socket_transport(a, asyncio.Protocol())
        f = asyncio.ensure_future(self.loop.sock_recv(b, 100), loop=self.loop)
        trans.close()
        self.loop.run_until_complete(f)
        self.assertEqual(f.result(), b'')
        b.close()

    def test_double_bind(self):
        ADDRESS = r'\\.\pipe\test_double_bind-%s' % os.getpid()
        server1 = windows_events.PipeServer(ADDRESS)
        with self.assertRaises(PermissionError):
            windows_events.PipeServer(ADDRESS)
        server1.close()

    def test_pipe(self):
        res = self.loop.run_until_complete(self._test_pipe())
        self.assertEqual(res, 'done')

    async def _test_pipe(self):
        ADDRESS = r'\\.\pipe\_test_pipe-%s' % os.getpid()

        with self.assertRaises(FileNotFoundError):
            await self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        [server] = await self.loop.start_serving_pipe(
            UpperProto, ADDRESS)
        self.assertIsInstance(server, windows_events.PipeServer)

        clients = []
        for i in range(5):
            stream_reader = asyncio.StreamReader(loop=self.loop)
            protocol = asyncio.StreamReaderProtocol(stream_reader,
                                                    loop=self.loop)
            trans, proto = await self.loop.create_pipe_connection(
                lambda: protocol, ADDRESS)
            self.assertIsInstance(trans, asyncio.Transport)
            self.assertEqual(protocol, proto)
            clients.append((stream_reader, trans))

        for i, (r, w) in enumerate(clients):
            w.write('lower-{}\n'.format(i).encode())

        for i, (r, w) in enumerate(clients):
            response = await r.readline()
            self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
            w.close()

        server.close()

        with self.assertRaises(FileNotFoundError):
            await self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        return 'done'

    def test_connect_pipe_cancel(self):
        exc = OSError()
        exc.winerror = _overlapped.ERROR_PIPE_BUSY
        with mock.patch.object(_overlapped, 'ConnectPipe',
                               side_effect=exc) as connect:
            coro = self.loop._proactor.connect_pipe('pipe_address')
            task = self.loop.create_task(coro)

            
            task.cancel()
            with self.assertRaises(asyncio.CancelledError):
                self.loop.run_until_complete(task)

    def test_wait_for_handle(self):
        event = _overlapped.CreateEvent(None, True, False, None)
        self.addCleanup(_winapi.CloseHandle, event)

        
        
        timeout = 0.5
        fut = self.loop._proactor.wait_for_handle(event, timeout)
        start = self.loop.time()
        done = self.loop.run_until_complete(fut)
        elapsed = self.loop.time() - start

        self.assertEqual(done, False)
        self.assertFalse(fut.result())
        self.assertGreaterEqual(elapsed, timeout - test_utils.CLOCK_RES)

        _overlapped.SetEvent(event)

        
        
        fut = self.loop._proactor.wait_for_handle(event, 10)
        done = self.loop.run_until_complete(fut)

        self.assertEqual(done, True)
        self.assertTrue(fut.result())

        
        
        fut.cancel()

    def test_wait_for_handle_cancel(self):
        event = _overlapped.CreateEvent(None, True, False, None)
        self.addCleanup(_winapi.CloseHandle, event)

        
        
        fut = self.loop._proactor.wait_for_handle(event, 10)
        fut.cancel()
        with self.assertRaises(asyncio.CancelledError):
            self.loop.run_until_complete(fut)

        
        
        fut = self.loop._proactor.wait_for_handle(event)
        fut.cancel()
        fut.cancel()

    def test_read_self_pipe_restart(self):
        
        
        
        self.loop.call_exception_handler = mock.Mock()
        
        
        
        
        f = self.loop.run_in_executor(None, lambda: None)
        self.loop.stop()
        self.loop.run_forever()
        self.loop.stop()
        self.loop.run_forever()

        
        
        
        
        
        
        
        self.loop.run_until_complete(f)
        
        
        self.close_loop(self.loop)
        self.assertFalse(self.loop.call_exception_handler.called)

    def test_address_argument_type_error(self):
        
        proactor = self.loop._proactor
        sock = socket.socket(type=socket.SOCK_DGRAM)
        bad_address = None
        with self.assertRaises(TypeError):
            proactor.connect(sock, bad_address)
        with self.assertRaises(TypeError):
            proactor.sendto(sock, b'abc', addr=bad_address)
        sock.close()

    def test_client_pipe_stat(self):
        res = self.loop.run_until_complete(self._test_client_pipe_stat())
        self.assertEqual(res, 'done')

    async def _test_client_pipe_stat(self):
        
        ADDRESS = r'\\.\pipe\test_client_pipe_stat-%s' % os.getpid()

        async def probe():
            
            h = _overlapped.ConnectPipe(ADDRESS)
            try:
                _winapi.CloseHandle(_overlapped.ConnectPipe(ADDRESS))
            except OSError as e:
                if e.winerror != _overlapped.ERROR_PIPE_BUSY:
                    raise
            finally:
                _winapi.CloseHandle(h)

        with self.assertRaises(FileNotFoundError):
            await probe()

        [server] = await self.loop.start_serving_pipe(asyncio.Protocol, ADDRESS)
        self.assertIsInstance(server, windows_events.PipeServer)

        errors = []
        self.loop.set_exception_handler(lambda _, data: errors.append(data))

        for i in range(5):
            await self.loop.create_task(probe())

        self.assertEqual(len(errors), 0, errors)

        server.close()

        with self.assertRaises(FileNotFoundError):
            await probe()

        return "done"


class WinPolicyTests(test_utils.TestCase):

    def test_selector_win_policy(self):
        async def main():
            self.assertIsInstance(
                asyncio.get_running_loop(),
                asyncio.SelectorEventLoop)

        old_policy = asyncio.get_event_loop_policy()
        try:
            asyncio.set_event_loop_policy(
                asyncio.WindowsSelectorEventLoopPolicy())
            asyncio.run(main())
        finally:
            asyncio.set_event_loop_policy(old_policy)

    def test_proactor_win_policy(self):
        async def main():
            self.assertIsInstance(
                asyncio.get_running_loop(),
                asyncio.ProactorEventLoop)

        old_policy = asyncio.get_event_loop_policy()
        try:
            asyncio.set_event_loop_policy(
                asyncio.WindowsProactorEventLoopPolicy())
            asyncio.run(main())
        finally:
            asyncio.set_event_loop_policy(old_policy)


if __name__ == '__main__':
    unittest.main()