1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
|
import os
import sys
import unittest
if sys.platform != 'win32':
raise unittest.SkipTest('Windows only')
import _winapi
import asyncio
from asyncio import _overlapped
from asyncio import windows_events
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 ProactorTests(unittest.TestCase):
def setUp(self):
self.loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(None)
def tearDown(self):
self.loop.close()
self.loop = None
def test_close(self):
a, b = self.loop._socketpair()
trans = self.loop._make_socket_transport(a, asyncio.Protocol())
f = asyncio.async(self.loop.sock_recv(b, 100))
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')
def _test_pipe(self):
ADDRESS = r'\\.\pipe\_test_pipe-%s' % os.getpid()
with self.assertRaises(FileNotFoundError):
yield from self.loop.create_pipe_connection(
asyncio.Protocol, ADDRESS)
[server] = yield from 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)
trans, proto = yield from 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 = yield from r.readline()
self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
w.close()
server.close()
with self.assertRaises(FileNotFoundError):
yield from self.loop.create_pipe_connection(
asyncio.Protocol, ADDRESS)
return 'done'
def test_wait_for_handle(self):
event = _overlapped.CreateEvent(None, True, False, None)
self.addCleanup(_winapi.CloseHandle, event)
# Wait for unset event with 0.2s timeout;
# result should be False at timeout
f = self.loop._proactor.wait_for_handle(event, 0.2)
start = self.loop.time()
self.loop.run_until_complete(f)
elapsed = self.loop.time() - start
self.assertFalse(f.result())
self.assertTrue(0.18 < elapsed < 0.9, elapsed)
_overlapped.SetEvent(event)
# Wait for for set event;
# result should be True immediately
f = self.loop._proactor.wait_for_handle(event, 10)
start = self.loop.time()
self.loop.run_until_complete(f)
elapsed = self.loop.time() - start
self.assertTrue(f.result())
self.assertTrue(0 <= elapsed < 0.1, elapsed)
_overlapped.ResetEvent(event)
# Wait for unset event with a cancelled future;
# CancelledError should be raised immediately
f = self.loop._proactor.wait_for_handle(event, 10)
f.cancel()
start = self.loop.time()
with self.assertRaises(asyncio.CancelledError):
self.loop.run_until_complete(f)
elapsed = self.loop.time() - start
self.assertTrue(0 <= elapsed < 0.1, elapsed)
if __name__ == '__main__':
unittest.main()
|
