(
self,
event: str,
predicate: Callable = None,
timeout: float = None,
)
| 976 | return self._initializer["url"] |
| 977 | |
| 978 | def expect_event( |
| 979 | self, |
| 980 | event: str, |
| 981 | predicate: Callable = None, |
| 982 | timeout: float = None, |
| 983 | ) -> EventContextManagerImpl: |
| 984 | if timeout is None: |
| 985 | timeout = cast(Any, self._parent)._timeout_settings.timeout() |
| 986 | waiter = Waiter(self, f"web_socket.expect_event({event})") |
| 987 | waiter.reject_on_timeout( |
| 988 | cast(float, timeout), |
| 989 | f'Timeout {timeout}ms exceeded while waiting for event "{event}"', |
| 990 | ) |
| 991 | if event != WebSocket.Events.Close: |
| 992 | waiter.reject_on_event(self, WebSocket.Events.Close, Error("Socket closed")) |
| 993 | if event != WebSocket.Events.Error: |
| 994 | waiter.reject_on_event(self, WebSocket.Events.Error, Error("Socket error")) |
| 995 | waiter.reject_on_event( |
| 996 | self._page, "close", lambda: self._page._close_error_with_reason() |
| 997 | ) |
| 998 | waiter.wait_for_event(self, event, predicate) |
| 999 | return EventContextManagerImpl(waiter.result()) |
| 1000 | |
| 1001 | async def wait_for_event( |
| 1002 | self, event: str, predicate: Callable = None, timeout: float = None |
no test coverage detected