| 203 | self.assertEqual(test_proc.exitcode, errno.ETIMEDOUT) |
| 204 | |
| 205 | def test_multiple_interrupts(self): |
| 206 | timed_out = multiprocessing.Event() |
| 207 | all_read = multiprocessing.Event() |
| 208 | test_proc = multiprocessing.Process(target=self.socket_fn, args=(timed_out, all_read)) |
| 209 | test_proc.daemon = True |
| 210 | test_proc.start() |
| 211 | while not all_read.is_set(): |
| 212 | time.sleep(1.0) |
| 213 | while not timed_out.is_set(): |
| 214 | os.kill(test_proc.pid, signal.SIGWINCH) |
| 215 | time.sleep(1.0) |
| 216 | test_proc.join(timeout=5.0) |
| 217 | self.assertEqual(test_proc.exitcode, errno.ETIMEDOUT) |
| 218 | |
| 219 | def test_maxread(self): |
| 220 | sock = socket.socket(self.af, socket.SOCK_STREAM) |