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)
| 154 | self.check_reentrant_write("xy", mode="w", encoding="ascii") |
| 155 | |
| 156 | def check_interrupted_read_retry(self, decode, **fdopen_kwargs): |
| 157 | """Check that a buffered read, when it gets interrupted (either |
| 158 | returning a partial result or EINTR), properly invokes the signal |
| 159 | handler and retries if the latter returned successfully.""" |
| 160 | r, w = os.pipe() |
| 161 | fdopen_kwargs["closefd"] = False |
| 162 | def alarm_handler(sig, frame): |
| 163 | os.write(w, b"bar") |
| 164 | signal.signal(signal.SIGALRM, alarm_handler) |
| 165 | try: |
| 166 | rio = self.io.open(r, **fdopen_kwargs) |
| 167 | os.write(w, b"foo") |
| 168 | signal.alarm(1) |
| 169 | # Expected behaviour: |
| 170 | # - first raw read() returns partial b"foo" |
| 171 | # - second raw read() returns EINTR |
| 172 | # - third raw read() returns b"bar" |
| 173 | self.assertEqual(decode(rio.read(6)), "foobar") |
| 174 | finally: |
| 175 | signal.alarm(0) |
| 176 | rio.close() |
| 177 | os.close(w) |
| 178 | os.close(r) |
| 179 | |
| 180 | @requires_alarm |
| 181 | @support.requires_resource('walltime') |
no test coverage detected