(self)
| 3237 | |
| 3238 | @contextmanager |
| 3239 | def _sigint_handler(self): |
| 3240 | # Signal handling strategy: |
| 3241 | # - When we call input() we want a SIGINT to raise KeyboardInterrupt |
| 3242 | # - Otherwise we want to write to the wakeup FD and set a flag. |
| 3243 | # We'll break out of select() when the wakeup FD is written to, |
| 3244 | # and we'll check the flag whenever we're about to accept input. |
| 3245 | def handler(signum, frame): |
| 3246 | self.sigint_received = True |
| 3247 | if self.raise_on_sigint: |
| 3248 | # One-shot; don't raise again until the flag is set again. |
| 3249 | self.raise_on_sigint = False |
| 3250 | self.sigint_received = False |
| 3251 | raise KeyboardInterrupt |
| 3252 | |
| 3253 | sentinel = object() |
| 3254 | old_handler = sentinel |
| 3255 | old_wakeup_fd = sentinel |
| 3256 | |
| 3257 | self.signal_read, self.signal_write = socket.socketpair() |
| 3258 | with (closing(self.signal_read), closing(self.signal_write)): |
| 3259 | self.signal_read.setblocking(False) |
| 3260 | self.signal_write.setblocking(False) |
| 3261 | |
| 3262 | try: |
| 3263 | old_handler = signal.signal(signal.SIGINT, handler) |
| 3264 | |
| 3265 | try: |
| 3266 | old_wakeup_fd = signal.set_wakeup_fd( |
| 3267 | self.signal_write.fileno(), |
| 3268 | warn_on_full_buffer=False, |
| 3269 | ) |
| 3270 | yield |
| 3271 | finally: |
| 3272 | # Restore the old wakeup fd if we installed a new one |
| 3273 | if old_wakeup_fd is not sentinel: |
| 3274 | signal.set_wakeup_fd(old_wakeup_fd) |
| 3275 | finally: |
| 3276 | self.signal_read = self.signal_write = None |
| 3277 | if old_handler is not sentinel: |
| 3278 | # Restore the old handler if we installed a new one |
| 3279 | signal.signal(signal.SIGINT, old_handler) |
| 3280 | |
| 3281 | @contextmanager |
| 3282 | def _sigint_raises_keyboard_interrupt(self): |
no test coverage detected