(self)
| 126 | self.assertIsNotNone(re.match(r".\:\\PROGRA~\d", actual.upper()), actual) |
| 127 | |
| 128 | def test_namedpipe(self): |
| 129 | pipe_name = rf"\\.\pipe\LOCAL\{os_helper.TESTFN}" |
| 130 | |
| 131 | # Pipe does not exist, so this raises |
| 132 | with self.assertRaises(FileNotFoundError): |
| 133 | _winapi.WaitNamedPipe(pipe_name, 0) |
| 134 | |
| 135 | pipe = _winapi.CreateNamedPipe( |
| 136 | pipe_name, |
| 137 | _winapi.PIPE_ACCESS_DUPLEX, |
| 138 | 8, # 8=PIPE_REJECT_REMOTE_CLIENTS |
| 139 | 2, # two instances available |
| 140 | 32, 32, 0, 0) |
| 141 | self.addCleanup(_winapi.CloseHandle, pipe) |
| 142 | |
| 143 | # Pipe instance is available, so this passes |
| 144 | _winapi.WaitNamedPipe(pipe_name, 0) |
| 145 | |
| 146 | with open(pipe_name, 'w+b') as pipe2: |
| 147 | # No instances available, so this times out |
| 148 | # (WinError 121 does not get mapped to TimeoutError) |
| 149 | with self.assertRaises(OSError): |
| 150 | _winapi.WaitNamedPipe(pipe_name, 0) |
| 151 | |
| 152 | _winapi.WriteFile(pipe, b'testdata') |
| 153 | self.assertEqual(b'testdata', pipe2.read(8)) |
| 154 | |
| 155 | self.assertEqual((b'', 0), _winapi.PeekNamedPipe(pipe, 8)[:2]) |
| 156 | pipe2.write(b'testdata') |
| 157 | pipe2.flush() |
| 158 | self.assertEqual((b'testdata', 8), _winapi.PeekNamedPipe(pipe, 8)[:2]) |
nothing calls this directly
no test coverage detected