(self)
| 81 | return f"<Timeout [{self._state.value}]{info_str}>" |
| 82 | |
| 83 | async def __aenter__(self) -> "Timeout": |
| 84 | if self._state is not _State.CREATED: |
| 85 | raise RuntimeError("Timeout has already been entered") |
| 86 | task = tasks.current_task() |
| 87 | if task is None: |
| 88 | raise RuntimeError("Timeout should be used inside a task") |
| 89 | self._state = _State.ENTERED |
| 90 | self._task = task |
| 91 | self._cancelling = self._task.cancelling() |
| 92 | self.reschedule(self._when) |
| 93 | return self |
| 94 | |
| 95 | async def __aexit__( |
| 96 | self, |
nothing calls this directly
no test coverage detected