(self)
| 611 | |
| 612 | @enable_ki_protection |
| 613 | def __enter__(self) -> Self: |
| 614 | task = _core.current_task() |
| 615 | if self._has_been_entered: |
| 616 | raise RuntimeError( |
| 617 | "Each CancelScope may only be used for a single 'with' block", |
| 618 | ) |
| 619 | self._has_been_entered = True |
| 620 | |
| 621 | if self._relative_deadline != inf: |
| 622 | assert self._deadline == inf |
| 623 | self._deadline = current_time() + self._relative_deadline |
| 624 | self._relative_deadline = inf |
| 625 | |
| 626 | if current_time() >= self._deadline: |
| 627 | self._cancel(CancelReason(source="deadline")) |
| 628 | with self._might_change_registered_deadline(): |
| 629 | self._cancel_status = CancelStatus(scope=self, parent=task._cancel_status) |
| 630 | task._activate_cancel_status(self._cancel_status) |
| 631 | return self |
| 632 | |
| 633 | def _close(self, exc: BaseException | None) -> BaseException | None: |
| 634 | if self._cancel_status is None: |
no test coverage detected