Check that a partial write, when it gets interrupted, properly invokes the signal handler, and bubbles up the exception raised in the latter.
(self, item, bytes, **fdopen_kwargs)
| 26 | 1/0 |
| 27 | |
| 28 | def check_interrupted_write(self, item, bytes, **fdopen_kwargs): |
| 29 | """Check that a partial write, when it gets interrupted, properly |
| 30 | invokes the signal handler, and bubbles up the exception raised |
| 31 | in the latter.""" |
| 32 | |
| 33 | # XXX This test has three flaws that appear when objects are |
| 34 | # XXX not reference counted. |
| 35 | |
| 36 | # - if wio.write() happens to trigger a garbage collection, |
| 37 | # the signal exception may be raised when some __del__ |
| 38 | # method is running; it will not reach the assertRaises() |
| 39 | # call. |
| 40 | |
| 41 | # - more subtle, if the wio object is not destroyed at once |
| 42 | # and survives this function, the next opened file is likely |
| 43 | # to have the same fileno (since the file descriptor was |
| 44 | # actively closed). When wio.__del__ is finally called, it |
| 45 | # will close the other's test file... To trigger this with |
| 46 | # CPython, try adding "global wio" in this function. |
| 47 | |
| 48 | # - This happens only for streams created by the _pyio module, |
| 49 | # because a wio.close() that fails still consider that the |
| 50 | # file needs to be closed again. You can try adding an |
| 51 | # "assert wio.closed" at the end of the function. |
| 52 | |
| 53 | # Fortunately, a little gc.collect() seems to be enough to |
| 54 | # work around all these issues. |
| 55 | support.gc_collect() # For PyPy or other GCs. |
| 56 | |
| 57 | read_results = [] |
| 58 | def _read(): |
| 59 | s = os.read(r, 1) |
| 60 | read_results.append(s) |
| 61 | |
| 62 | t = threading.Thread(target=_read) |
| 63 | t.daemon = True |
| 64 | r, w = os.pipe() |
| 65 | fdopen_kwargs["closefd"] = False |
| 66 | large_data = item * (support.PIPE_MAX_SIZE // len(item) + 1) |
| 67 | try: |
| 68 | wio = self.io.open(w, **fdopen_kwargs) |
| 69 | if hasattr(signal, 'pthread_sigmask'): |
| 70 | # create the thread with SIGALRM signal blocked |
| 71 | signal.pthread_sigmask(signal.SIG_BLOCK, [signal.SIGALRM]) |
| 72 | t.start() |
| 73 | signal.pthread_sigmask(signal.SIG_UNBLOCK, [signal.SIGALRM]) |
| 74 | else: |
| 75 | t.start() |
| 76 | |
| 77 | # Fill the pipe enough that the write will be blocking. |
| 78 | # It will be interrupted by the timer armed above. Since the |
| 79 | # other thread has read one byte, the low-level write will |
| 80 | # return with a successful (partial) result rather than an EINTR. |
| 81 | # The buffered IO layer must check for pending signal |
| 82 | # handlers, which in this case will invoke alarm_interrupt(). |
| 83 | signal.alarm(1) |
| 84 | try: |
| 85 | self.assertRaises(ZeroDivisionError, wio.write, large_data) |
no test coverage detected