(self, wait_func)
| 408 | """ EINTR tests for the signal module. """ |
| 409 | |
| 410 | def check_sigwait(self, wait_func): |
| 411 | signum = signal.SIGUSR1 |
| 412 | |
| 413 | old_handler = signal.signal(signum, lambda *args: None) |
| 414 | self.addCleanup(signal.signal, signum, old_handler) |
| 415 | |
| 416 | code = '\n'.join(( |
| 417 | 'import os, time', |
| 418 | f'pid = {os.getpid()}', |
| 419 | f'signum = {int(signum)}', |
| 420 | f'sleep_time = {self.sleep_time!r}', |
| 421 | 'time.sleep(sleep_time)', |
| 422 | 'os.kill(pid, signum)', |
| 423 | )) |
| 424 | |
| 425 | signal.pthread_sigmask(signal.SIG_BLOCK, [signum]) |
| 426 | self.addCleanup(signal.pthread_sigmask, signal.SIG_UNBLOCK, [signum]) |
| 427 | |
| 428 | proc = self.subprocess(code) |
| 429 | with kill_on_error(proc): |
| 430 | wait_func(signum) |
| 431 | |
| 432 | self.assertEqual(proc.wait(), 0) |
| 433 | |
| 434 | @unittest.skipUnless(hasattr(signal, 'sigwaitinfo'), |
| 435 | 'need signal.sigwaitinfo()') |
no test coverage detected