Interrupt a subprocess after a second.
(self, command)
| 114 | self.assertEqual(status, 0) |
| 115 | |
| 116 | def assert_interrupts(self, command): |
| 117 | """ |
| 118 | Interrupt a subprocess after a second. |
| 119 | """ |
| 120 | if threading.main_thread() != threading.current_thread(): |
| 121 | raise nt.SkipTest("Can't run this test if not in main thread.") |
| 122 | |
| 123 | # Some tests can overwrite SIGINT handler (by using pdb for example), |
| 124 | # which then breaks this test, so just make sure it's operating |
| 125 | # normally. |
| 126 | signal.signal(signal.SIGINT, signal.default_int_handler) |
| 127 | |
| 128 | def interrupt(): |
| 129 | # Wait for subprocess to start: |
| 130 | time.sleep(0.5) |
| 131 | interrupt_main() |
| 132 | |
| 133 | threading.Thread(target=interrupt).start() |
| 134 | start = time.time() |
| 135 | try: |
| 136 | result = command() |
| 137 | except KeyboardInterrupt: |
| 138 | # Success! |
| 139 | pass |
| 140 | end = time.time() |
| 141 | self.assertTrue( |
| 142 | end - start < 2, "Process didn't die quickly: %s" % (end - start) |
| 143 | ) |
| 144 | return result |
| 145 | |
| 146 | def test_system_interrupt(self): |
| 147 | """ |
no test coverage detected