(self, sig)
| 23 | |
| 24 | class Win32KillTests(unittest.TestCase): |
| 25 | def _kill(self, sig): |
| 26 | # Start sys.executable as a subprocess and communicate from the |
| 27 | # subprocess to the parent that the interpreter is ready. When it |
| 28 | # becomes ready, send *sig* via os.kill to the subprocess and check |
| 29 | # that the return code is equal to *sig*. |
| 30 | import ctypes |
| 31 | from ctypes import wintypes |
| 32 | import msvcrt |
| 33 | |
| 34 | # Since we can't access the contents of the process' stdout until the |
| 35 | # process has exited, use PeekNamedPipe to see what's inside stdout |
| 36 | # without waiting. This is done so we can tell that the interpreter |
| 37 | # is started and running at a point where it could handle a signal. |
| 38 | PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe |
| 39 | PeekNamedPipe.restype = wintypes.BOOL |
| 40 | PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle |
| 41 | ctypes.POINTER(ctypes.c_char), # stdout buf |
| 42 | wintypes.DWORD, # Buffer size |
| 43 | ctypes.POINTER(wintypes.DWORD), # bytes read |
| 44 | ctypes.POINTER(wintypes.DWORD), # bytes avail |
| 45 | ctypes.POINTER(wintypes.DWORD)) # bytes left |
| 46 | msg = "running" |
| 47 | proc = subprocess.Popen([sys.executable, "-c", |
| 48 | "import sys;" |
| 49 | "sys.stdout.write('{}');" |
| 50 | "sys.stdout.flush();" |
| 51 | "input()".format(msg)], |
| 52 | stdout=subprocess.PIPE, |
| 53 | stderr=subprocess.PIPE, |
| 54 | stdin=subprocess.PIPE) |
| 55 | self.addCleanup(proc.stdout.close) |
| 56 | self.addCleanup(proc.stderr.close) |
| 57 | self.addCleanup(proc.stdin.close) |
| 58 | |
| 59 | count, max = 0, 100 |
| 60 | while count < max and proc.poll() is None: |
| 61 | # Create a string buffer to store the result of stdout from the pipe |
| 62 | buf = ctypes.create_string_buffer(len(msg)) |
| 63 | # Obtain the text currently in proc.stdout |
| 64 | # Bytes read/avail/left are left as NULL and unused |
| 65 | rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()), |
| 66 | buf, ctypes.sizeof(buf), None, None, None) |
| 67 | self.assertNotEqual(rslt, 0, "PeekNamedPipe failed") |
| 68 | if buf.value: |
| 69 | self.assertEqual(msg, buf.value.decode()) |
| 70 | break |
| 71 | time.sleep(0.1) |
| 72 | count += 1 |
| 73 | else: |
| 74 | self.fail("Did not receive communication from the subprocess") |
| 75 | |
| 76 | os.kill(proc.pid, sig) |
| 77 | self.assertEqual(proc.wait(), sig) |
| 78 | |
| 79 | def test_kill_sigterm(self): |
| 80 | # SIGTERM doesn't mean anything special, but make sure it works |
no test coverage detected