(self, data, **fdopen_kwargs)
| 119 | |
| 120 | @support.no_tracing |
| 121 | def check_reentrant_write(self, data, **fdopen_kwargs): |
| 122 | def on_alarm(*args): |
| 123 | # Will be called reentrantly from the same thread |
| 124 | wio.write(data) |
| 125 | 1/0 |
| 126 | signal.signal(signal.SIGALRM, on_alarm) |
| 127 | r, w = os.pipe() |
| 128 | wio = self.io.open(w, **fdopen_kwargs) |
| 129 | try: |
| 130 | signal.alarm(1) |
| 131 | # Either the reentrant call to wio.write() fails with RuntimeError, |
| 132 | # or the signal handler raises ZeroDivisionError. |
| 133 | with self.assertRaises((ZeroDivisionError, RuntimeError)) as cm: |
| 134 | while 1: |
| 135 | for i in range(100): |
| 136 | wio.write(data) |
| 137 | wio.flush() |
| 138 | # Make sure the buffer doesn't fill up and block further writes |
| 139 | os.read(r, len(data) * 100) |
| 140 | exc = cm.exception |
| 141 | if isinstance(exc, RuntimeError): |
| 142 | self.assertStartsWith(str(exc), "reentrant call") |
| 143 | finally: |
| 144 | signal.alarm(0) |
| 145 | wio.close() |
| 146 | os.close(r) |
| 147 | |
| 148 | @requires_alarm |
| 149 | def test_reentrant_write_buffered(self): |
no test coverage detected