Check that a buffered read, when it gets interrupted (either returning a partial result or EINTR), properly invokes the signal handler and retries if the latter returned successfully.
(self, decode, **fdopen_kwargs)
| 4968 | self.check_reentrant_write("xy", mode="w", encoding="ascii") |
| 4969 | |
| 4970 | def check_interrupted_read_retry(self, decode, **fdopen_kwargs): |
| 4971 | """Check that a buffered read, when it gets interrupted (either |
| 4972 | returning a partial result or EINTR), properly invokes the signal |
| 4973 | handler and retries if the latter returned successfully.""" |
| 4974 | r, w = os.pipe() |
| 4975 | fdopen_kwargs["closefd"] = False |
| 4976 | def alarm_handler(sig, frame): |
| 4977 | os.write(w, b"bar") |
| 4978 | signal.signal(signal.SIGALRM, alarm_handler) |
| 4979 | try: |
| 4980 | rio = self.io.open(r, **fdopen_kwargs) |
| 4981 | os.write(w, b"foo") |
| 4982 | signal.alarm(1) |
| 4983 | # Expected behaviour: |
| 4984 | # - first raw read() returns partial b"foo" |
| 4985 | # - second raw read() returns EINTR |
| 4986 | # - third raw read() returns b"bar" |
| 4987 | self.assertEqual(decode(rio.read(6)), "foobar") |
| 4988 | finally: |
| 4989 | signal.alarm(0) |
| 4990 | rio.close() |
| 4991 | os.close(w) |
| 4992 | os.close(r) |
| 4993 | |
| 4994 | @requires_alarm |
| 4995 | @support.requires_resource('walltime') |
no test coverage detected