| 189 | session.expect(b'Bogus response') |
| 190 | |
| 191 | def test_interrupt(self): |
| 192 | timed_out = multiprocessing.Event() |
| 193 | all_read = multiprocessing.Event() |
| 194 | test_proc = multiprocessing.Process(target=self.socket_fn, args=(timed_out, all_read)) |
| 195 | test_proc.daemon = True |
| 196 | test_proc.start() |
| 197 | while not all_read.is_set(): |
| 198 | time.sleep(1.0) |
| 199 | os.kill(test_proc.pid, signal.SIGWINCH) |
| 200 | while not timed_out.is_set(): |
| 201 | time.sleep(1.0) |
| 202 | test_proc.join(timeout=5.0) |
| 203 | self.assertEqual(test_proc.exitcode, errno.ETIMEDOUT) |
| 204 | |
| 205 | def test_multiple_interrupts(self): |
| 206 | timed_out = multiprocessing.Event() |