(self)
| 254 | self._timeout_handler = self._loop.call_at(deadline, self._on_timeout) |
| 255 | |
| 256 | def _do_enter(self) -> None: |
| 257 | if self._state != _State.INIT: |
| 258 | raise RuntimeError(f"invalid state {self._state.value}") |
| 259 | self._state = _State.ENTER |
| 260 | self._reschedule() |
| 261 | |
| 262 | def _do_exit(self, exc_type: Optional[Type[BaseException]]) -> None: |
| 263 | if exc_type is asyncio.CancelledError and self._state == _State.TIMEOUT: |
no test coverage detected