(self, sig)
| 2805 | @unittest.skipUnless(sys.platform == "win32", "Win32 specific tests") |
| 2806 | class Win32KillTests(unittest.TestCase): |
| 2807 | def _kill(self, sig): |
| 2808 | # Start sys.executable as a subprocess and communicate from the |
| 2809 | # subprocess to the parent that the interpreter is ready. When it |
| 2810 | # becomes ready, send *sig* via os.kill to the subprocess and check |
| 2811 | # that the return code is equal to *sig*. |
| 2812 | import ctypes |
| 2813 | from ctypes import wintypes |
| 2814 | import msvcrt |
| 2815 | |
| 2816 | # Since we can't access the contents of the process' stdout until the |
| 2817 | # process has exited, use PeekNamedPipe to see what's inside stdout |
| 2818 | # without waiting. This is done so we can tell that the interpreter |
| 2819 | # is started and running at a point where it could handle a signal. |
| 2820 | PeekNamedPipe = ctypes.windll.kernel32.PeekNamedPipe |
| 2821 | PeekNamedPipe.restype = wintypes.BOOL |
| 2822 | PeekNamedPipe.argtypes = (wintypes.HANDLE, # Pipe handle |
| 2823 | ctypes.POINTER(ctypes.c_char), # stdout buf |
| 2824 | wintypes.DWORD, # Buffer size |
| 2825 | ctypes.POINTER(wintypes.DWORD), # bytes read |
| 2826 | ctypes.POINTER(wintypes.DWORD), # bytes avail |
| 2827 | ctypes.POINTER(wintypes.DWORD)) # bytes left |
| 2828 | msg = "running" |
| 2829 | proc = subprocess.Popen([sys.executable, "-c", |
| 2830 | "import sys;" |
| 2831 | "sys.stdout.write('{}');" |
| 2832 | "sys.stdout.flush();" |
| 2833 | "input()".format(msg)], |
| 2834 | stdout=subprocess.PIPE, |
| 2835 | stderr=subprocess.PIPE, |
| 2836 | stdin=subprocess.PIPE) |
| 2837 | self.addCleanup(proc.stdout.close) |
| 2838 | self.addCleanup(proc.stderr.close) |
| 2839 | self.addCleanup(proc.stdin.close) |
| 2840 | |
| 2841 | count, max = 0, 100 |
| 2842 | while count < max and proc.poll() is None: |
| 2843 | # Create a string buffer to store the result of stdout from the pipe |
| 2844 | buf = ctypes.create_string_buffer(len(msg)) |
| 2845 | # Obtain the text currently in proc.stdout |
| 2846 | # Bytes read/avail/left are left as NULL and unused |
| 2847 | rslt = PeekNamedPipe(msvcrt.get_osfhandle(proc.stdout.fileno()), |
| 2848 | buf, ctypes.sizeof(buf), None, None, None) |
| 2849 | self.assertNotEqual(rslt, 0, "PeekNamedPipe failed") |
| 2850 | if buf.value: |
| 2851 | self.assertEqual(msg, buf.value.decode()) |
| 2852 | break |
| 2853 | time.sleep(0.1) |
| 2854 | count += 1 |
| 2855 | else: |
| 2856 | self.fail("Did not receive communication from the subprocess") |
| 2857 | |
| 2858 | os.kill(proc.pid, sig) |
| 2859 | self.assertEqual(proc.wait(), sig) |
| 2860 | |
| 2861 | def test_kill_sigterm(self): |
| 2862 | # SIGTERM doesn't mean anything special, but make sure it works |
no test coverage detected