Fake a SIGINT happening during Popen._communicate() and ._wait(). This avoids the need to actually try and get test environments to send and receive signals reliably across platforms. The net effect of a ^C happening during a blocking subprocess execution which we want to c
(self, popener, mock__communicate,
**kwargs)
| 3897 | |
| 3898 | @mock.patch.object(subprocess.Popen, "_communicate") |
| 3899 | def _test_keyboardinterrupt_no_kill(self, popener, mock__communicate, |
| 3900 | **kwargs): |
| 3901 | """Fake a SIGINT happening during Popen._communicate() and ._wait(). |
| 3902 | |
| 3903 | This avoids the need to actually try and get test environments to send |
| 3904 | and receive signals reliably across platforms. The net effect of a ^C |
| 3905 | happening during a blocking subprocess execution which we want to clean |
| 3906 | up from is a KeyboardInterrupt coming out of communicate() or wait(). |
| 3907 | """ |
| 3908 | |
| 3909 | mock__communicate.side_effect = KeyboardInterrupt |
| 3910 | try: |
| 3911 | with mock.patch.object(subprocess.Popen, "_wait") as mock__wait: |
| 3912 | # We patch out _wait() as no signal was involved so the |
| 3913 | # child process isn't actually going to exit rapidly. |
| 3914 | mock__wait.side_effect = KeyboardInterrupt |
| 3915 | with mock.patch.object(subprocess, "Popen", |
| 3916 | self.RecordingPopen): |
| 3917 | with self.assertRaises(KeyboardInterrupt): |
| 3918 | popener([sys.executable, "-c", |
| 3919 | "import time\ntime.sleep(9)\nimport sys\n" |
| 3920 | "sys.stderr.write('\\n!runaway child!\\n')"], |
| 3921 | stdout=subprocess.DEVNULL, **kwargs) |
| 3922 | for call in mock__wait.call_args_list[1:]: |
| 3923 | self.assertNotEqual( |
| 3924 | call, mock.call(timeout=None), |
| 3925 | "no open-ended wait() after the first allowed: " |
| 3926 | f"{mock__wait.call_args_list}") |
| 3927 | sigint_calls = [] |
| 3928 | for call in mock__wait.call_args_list: |
| 3929 | if call == mock.call(timeout=0.25): # from Popen.__init__ |
| 3930 | sigint_calls.append(call) |
| 3931 | self.assertLessEqual(mock__wait.call_count, 2, |
| 3932 | msg=mock__wait.call_args_list) |
| 3933 | self.assertEqual(len(sigint_calls), 1, |
| 3934 | msg=mock__wait.call_args_list) |
| 3935 | finally: |
| 3936 | # cleanup the forgotten (due to our mocks) child process |
| 3937 | process = self.RecordingPopen.instances_created.pop() |
| 3938 | process.kill() |
| 3939 | process.wait() |
| 3940 | self.assertEqual([], self.RecordingPopen.instances_created) |
| 3941 | |
| 3942 | def test_call_keyboardinterrupt_no_kill(self): |
| 3943 | self._test_keyboardinterrupt_no_kill(subprocess.call, timeout=6.282) |
no test coverage detected