• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1# Test the Windows-only _winapi module
2
3import os
4import pathlib
5import random
6import re
7import threading
8import time
9import unittest
10from test.support import import_helper, os_helper
11
12_winapi = import_helper.import_module('_winapi', required_on=['win'])
13
14MAXIMUM_WAIT_OBJECTS = 64
15MAXIMUM_BATCHED_WAIT_OBJECTS = (MAXIMUM_WAIT_OBJECTS - 1) ** 2
16
17class WinAPIBatchedWaitForMultipleObjectsTests(unittest.TestCase):
18    def _events_waitall_test(self, n):
19        evts = [_winapi.CreateEventW(0, False, False, None) for _ in range(n)]
20
21        with self.assertRaises(TimeoutError):
22            _winapi.BatchedWaitForMultipleObjects(evts, True, 100)
23
24        # Ensure no errors raised when all are triggered
25        for e in evts:
26            _winapi.SetEvent(e)
27        try:
28            _winapi.BatchedWaitForMultipleObjects(evts, True, 100)
29        except TimeoutError:
30            self.fail("expected wait to complete immediately")
31
32        # Choose 8 events to set, distributed throughout the list, to make sure
33        # we don't always have them in the first chunk
34        chosen = [i * (len(evts) // 8) for i in range(8)]
35
36        # Replace events with invalid handles to make sure we fail
37        for i in chosen:
38            old_evt = evts[i]
39            evts[i] = -1
40            with self.assertRaises(OSError):
41                _winapi.BatchedWaitForMultipleObjects(evts, True, 100)
42            evts[i] = old_evt
43
44
45    def _events_waitany_test(self, n):
46        evts = [_winapi.CreateEventW(0, False, False, None) for _ in range(n)]
47
48        with self.assertRaises(TimeoutError):
49            _winapi.BatchedWaitForMultipleObjects(evts, False, 100)
50
51        # Choose 8 events to set, distributed throughout the list, to make sure
52        # we don't always have them in the first chunk
53        chosen = [i * (len(evts) // 8) for i in range(8)]
54
55        # Trigger one by one. They are auto-reset events, so will only trigger once
56        for i in chosen:
57            with self.subTest(f"trigger event {i} of {len(evts)}"):
58                _winapi.SetEvent(evts[i])
59                triggered = _winapi.BatchedWaitForMultipleObjects(evts, False, 10000)
60                self.assertSetEqual(set(triggered), {i})
61
62        # Trigger all at once. This may require multiple calls
63        for i in chosen:
64            _winapi.SetEvent(evts[i])
65        triggered = set()
66        while len(triggered) < len(chosen):
67            triggered.update(_winapi.BatchedWaitForMultipleObjects(evts, False, 10000))
68        self.assertSetEqual(triggered, set(chosen))
69
70        # Replace events with invalid handles to make sure we fail
71        for i in chosen:
72            with self.subTest(f"corrupt event {i} of {len(evts)}"):
73                old_evt = evts[i]
74                evts[i] = -1
75                with self.assertRaises(OSError):
76                    _winapi.BatchedWaitForMultipleObjects(evts, False, 100)
77                evts[i] = old_evt
78
79
80    def test_few_events_waitall(self):
81        self._events_waitall_test(16)
82
83    def test_many_events_waitall(self):
84        self._events_waitall_test(256)
85
86    def test_max_events_waitall(self):
87        self._events_waitall_test(MAXIMUM_BATCHED_WAIT_OBJECTS)
88
89
90    def test_few_events_waitany(self):
91        self._events_waitany_test(16)
92
93    def test_many_events_waitany(self):
94        self._events_waitany_test(256)
95
96    def test_max_events_waitany(self):
97        self._events_waitany_test(MAXIMUM_BATCHED_WAIT_OBJECTS)
98
99
100class WinAPITests(unittest.TestCase):
101    def test_getlongpathname(self):
102        testfn = pathlib.Path(os.getenv("ProgramFiles")).parents[-1] / "PROGRA~1"
103        if not os.path.isdir(testfn):
104            raise unittest.SkipTest("require x:\\PROGRA~1 to test")
105
106        # pathlib.Path will be rejected - only str is accepted
107        with self.assertRaises(TypeError):
108            _winapi.GetLongPathName(testfn)
109
110        actual = _winapi.GetLongPathName(os.fsdecode(testfn))
111
112        # Can't assume that PROGRA~1 expands to any particular variation, so
113        # ensure it matches any one of them.
114        candidates = set(testfn.parent.glob("Progra*"))
115        self.assertIn(pathlib.Path(actual), candidates)
116
117    def test_getshortpathname(self):
118        testfn = pathlib.Path(os.getenv("ProgramFiles"))
119        if not os.path.isdir(testfn):
120            raise unittest.SkipTest("require '%ProgramFiles%' to test")
121
122        # pathlib.Path will be rejected - only str is accepted
123        with self.assertRaises(TypeError):
124            _winapi.GetShortPathName(testfn)
125
126        actual = _winapi.GetShortPathName(os.fsdecode(testfn))
127
128        # Should contain "PROGRA~" but we can't predict the number
129        self.assertIsNotNone(re.match(r".\:\\PROGRA~\d", actual.upper()), actual)
130
131    def test_namedpipe(self):
132        pipe_name = rf"\\.\pipe\LOCAL\{os_helper.TESTFN}"
133
134        # Pipe does not exist, so this raises
135        with self.assertRaises(FileNotFoundError):
136            _winapi.WaitNamedPipe(pipe_name, 0)
137
138        pipe = _winapi.CreateNamedPipe(
139            pipe_name,
140            _winapi.PIPE_ACCESS_DUPLEX,
141            8, # 8=PIPE_REJECT_REMOTE_CLIENTS
142            2, # two instances available
143            32, 32, 0, 0)
144        self.addCleanup(_winapi.CloseHandle, pipe)
145
146        # Pipe instance is available, so this passes
147        _winapi.WaitNamedPipe(pipe_name, 0)
148
149        with open(pipe_name, 'w+b') as pipe2:
150            # No instances available, so this times out
151            # (WinError 121 does not get mapped to TimeoutError)
152            with self.assertRaises(OSError):
153                _winapi.WaitNamedPipe(pipe_name, 0)
154
155            _winapi.WriteFile(pipe, b'testdata')
156            self.assertEqual(b'testdata', pipe2.read(8))
157
158            self.assertEqual((b'', 0), _winapi.PeekNamedPipe(pipe, 8)[:2])
159            pipe2.write(b'testdata')
160            pipe2.flush()
161            self.assertEqual((b'testdata', 8), _winapi.PeekNamedPipe(pipe, 8)[:2])
162