1 22
import os
2

3 22
import pytest
4

5 22
on_windows = os.name == "nt"
6
# Mark all the tests in this file as being windows-only
7 22
pytestmark = pytest.mark.skipif(not on_windows, reason="windows only")
8

9 22
from .._core.tests.tutil import slow
10 22
import trio
11 22
from .. import _core
12 22
from .. import _timeouts
13

14 22
if on_windows:
15 9
    from .._core._windows_cffi import ffi, kernel32
16 9
    from .._wait_for_object import (
17
        WaitForSingleObject,
18
        WaitForMultipleObjects_sync,
19
    )
20

21

22 22
async def test_WaitForMultipleObjects_sync():
23
    # This does a series of tests where we set/close the handle before
24
    # initiating the waiting for it.
25
    #
26
    # Note that closing the handle (not signaling) will cause the
27
    # *initiation* of a wait to return immediately. But closing a handle
28
    # that is already being waited on will not stop whatever is waiting
29
    # for it.
30

31
    # One handle
32 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
33 9
    kernel32.SetEvent(handle1)
34 9
    WaitForMultipleObjects_sync(handle1)
35 9
    kernel32.CloseHandle(handle1)
36 9
    print("test_WaitForMultipleObjects_sync one OK")
37

38
    # Two handles, signal first
39 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
40 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
41 9
    kernel32.SetEvent(handle1)
42 9
    WaitForMultipleObjects_sync(handle1, handle2)
43 9
    kernel32.CloseHandle(handle1)
44 9
    kernel32.CloseHandle(handle2)
45 9
    print("test_WaitForMultipleObjects_sync set first OK")
46

47
    # Two handles, signal second
48 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
49 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
50 9
    kernel32.SetEvent(handle2)
51 9
    WaitForMultipleObjects_sync(handle1, handle2)
52 9
    kernel32.CloseHandle(handle1)
53 9
    kernel32.CloseHandle(handle2)
54 9
    print("test_WaitForMultipleObjects_sync set second OK")
55

56
    # Two handles, close first
57 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
58 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
59 9
    kernel32.CloseHandle(handle1)
60 9
    with pytest.raises(OSError):
61 9
        WaitForMultipleObjects_sync(handle1, handle2)
62 9
    kernel32.CloseHandle(handle2)
63 9
    print("test_WaitForMultipleObjects_sync close first OK")
64

65
    # Two handles, close second
66 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
67 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
68 9
    kernel32.CloseHandle(handle2)
69 9
    with pytest.raises(OSError):
70 9
        WaitForMultipleObjects_sync(handle1, handle2)
71 9
    kernel32.CloseHandle(handle1)
72 9
    print("test_WaitForMultipleObjects_sync close second OK")
73

74

75 22
@slow
76 9
async def test_WaitForMultipleObjects_sync_slow():
77
    # This does a series of test in which the main thread sync-waits for
78
    # handles, while we spawn a thread to set the handles after a short while.
79

80 9
    TIMEOUT = 0.3
81

82
    # One handle
83 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
84 9
    t0 = _core.current_time()
85 9
    async with _core.open_nursery() as nursery:
86 9
        nursery.start_soon(
87
            trio.to_thread.run_sync, WaitForMultipleObjects_sync, handle1
88
        )
89 9
        await _timeouts.sleep(TIMEOUT)
90
        # If we would comment the line below, the above thread will be stuck,
91
        # and Trio won't exit this scope
92 9
        kernel32.SetEvent(handle1)
93 9
    t1 = _core.current_time()
94 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
95 9
    kernel32.CloseHandle(handle1)
96 9
    print("test_WaitForMultipleObjects_sync_slow one OK")
97

98
    # Two handles, signal first
99 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
100 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
101 9
    t0 = _core.current_time()
102 9
    async with _core.open_nursery() as nursery:
103 9
        nursery.start_soon(
104
            trio.to_thread.run_sync, WaitForMultipleObjects_sync, handle1, handle2
105
        )
106 9
        await _timeouts.sleep(TIMEOUT)
107 9
        kernel32.SetEvent(handle1)
108 9
    t1 = _core.current_time()
109 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
110 9
    kernel32.CloseHandle(handle1)
111 9
    kernel32.CloseHandle(handle2)
112 9
    print("test_WaitForMultipleObjects_sync_slow thread-set first OK")
113

114
    # Two handles, signal second
115 9
    handle1 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
116 9
    handle2 = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
117 9
    t0 = _core.current_time()
118 9
    async with _core.open_nursery() as nursery:
119 9
        nursery.start_soon(
120
            trio.to_thread.run_sync, WaitForMultipleObjects_sync, handle1, handle2
121
        )
122 9
        await _timeouts.sleep(TIMEOUT)
123 9
        kernel32.SetEvent(handle2)
124 9
    t1 = _core.current_time()
125 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
126 9
    kernel32.CloseHandle(handle1)
127 9
    kernel32.CloseHandle(handle2)
128 9
    print("test_WaitForMultipleObjects_sync_slow thread-set second OK")
129

130

131 22
async def test_WaitForSingleObject():
132
    # This does a series of test for setting/closing the handle before
133
    # initiating the wait.
134

135
    # Test already set
136 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
137 9
    kernel32.SetEvent(handle)
138 9
    await WaitForSingleObject(handle)  # should return at once
139 9
    kernel32.CloseHandle(handle)
140 9
    print("test_WaitForSingleObject already set OK")
141

142
    # Test already set, as int
143 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
144 9
    handle_int = int(ffi.cast("intptr_t", handle))
145 9
    kernel32.SetEvent(handle)
146 9
    await WaitForSingleObject(handle_int)  # should return at once
147 9
    kernel32.CloseHandle(handle)
148 9
    print("test_WaitForSingleObject already set OK")
149

150
    # Test already closed
151 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
152 9
    kernel32.CloseHandle(handle)
153 9
    with pytest.raises(OSError):
154 9
        await WaitForSingleObject(handle)  # should return at once
155 9
    print("test_WaitForSingleObject already closed OK")
156

157
    # Not a handle
158 9
    with pytest.raises(TypeError):
159 9
        await WaitForSingleObject("not a handle")  # Wrong type
160
    # with pytest.raises(OSError):
161
    #     await WaitForSingleObject(99)  # If you're unlucky, it actually IS a handle :(
162 9
    print("test_WaitForSingleObject not a handle OK")
163

164

165 22
@slow
166 9
async def test_WaitForSingleObject_slow():
167
    # This does a series of test for setting the handle in another task,
168
    # and cancelling the wait task.
169

170
    # Set the timeout used in the tests. We test the waiting time against
171
    # the timeout with a certain margin.
172 9
    TIMEOUT = 0.3
173

174 9
    async def signal_soon_async(handle):
175 9
        await _timeouts.sleep(TIMEOUT)
176 9
        kernel32.SetEvent(handle)
177

178
    # Test handle is SET after TIMEOUT in separate coroutine
179

180 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
181 9
    t0 = _core.current_time()
182

183 9
    async with _core.open_nursery() as nursery:
184 9
        nursery.start_soon(WaitForSingleObject, handle)
185 9
        nursery.start_soon(signal_soon_async, handle)
186

187 9
    kernel32.CloseHandle(handle)
188 9
    t1 = _core.current_time()
189 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
190 9
    print("test_WaitForSingleObject_slow set from task OK")
191

192
    # Test handle is SET after TIMEOUT in separate coroutine, as int
193

194 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
195 9
    handle_int = int(ffi.cast("intptr_t", handle))
196 9
    t0 = _core.current_time()
197

198 9
    async with _core.open_nursery() as nursery:
199 9
        nursery.start_soon(WaitForSingleObject, handle_int)
200 9
        nursery.start_soon(signal_soon_async, handle)
201

202 9
    kernel32.CloseHandle(handle)
203 9
    t1 = _core.current_time()
204 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
205 9
    print("test_WaitForSingleObject_slow set from task as int OK")
206

207
    # Test handle is CLOSED after 1 sec - NOPE see comment above
208

209
    # Test cancellation
210

211 9
    handle = kernel32.CreateEventA(ffi.NULL, True, False, ffi.NULL)
212 9
    t0 = _core.current_time()
213

214 9
    with _timeouts.move_on_after(TIMEOUT):
215 9
        await WaitForSingleObject(handle)
216

217 9
    kernel32.CloseHandle(handle)
218 9
    t1 = _core.current_time()
219 9
    assert TIMEOUT <= (t1 - t0) < 2.0 * TIMEOUT
220 9
    print("test_WaitForSingleObject_slow cancellation OK")

Read our documentation on viewing source code .

Loading