(
self,
url: URLMatch = None,
waitUntil: DocumentLoadState = None,
timeout: float = None,
)
| 226 | } |
| 227 | |
| 228 | def expect_navigation( |
| 229 | self, |
| 230 | url: URLMatch = None, |
| 231 | waitUntil: DocumentLoadState = None, |
| 232 | timeout: float = None, |
| 233 | ) -> EventContextManagerImpl[Response]: |
| 234 | assert self._page |
| 235 | if not waitUntil: |
| 236 | waitUntil = "load" |
| 237 | |
| 238 | if timeout is None: |
| 239 | timeout = self._page._timeout_settings.navigation_timeout() |
| 240 | deadline = monotonic_time() + timeout |
| 241 | waiter = self._setup_navigation_waiter("expect_navigation", timeout) |
| 242 | |
| 243 | to_url = f' to "{url}"' if url else "" |
| 244 | waiter.log(f"waiting for navigation{to_url} until '{waitUntil}'") |
| 245 | |
| 246 | def predicate(event: Any) -> bool: |
| 247 | # Any failed navigation results in a rejection. |
| 248 | if event.get("error"): |
| 249 | return True |
| 250 | waiter.log(f' navigated to "{event["url"]}"') |
| 251 | return url_matches( |
| 252 | cast("Page", self._page)._browser_context._base_url, |
| 253 | event["url"], |
| 254 | url, |
| 255 | ) |
| 256 | |
| 257 | waiter.wait_for_event( |
| 258 | self._event_emitter, |
| 259 | "navigated", |
| 260 | predicate=predicate, |
| 261 | ) |
| 262 | |
| 263 | async def continuation() -> Optional[Response]: |
| 264 | event = await waiter.result() |
| 265 | if "error" in event: |
| 266 | raise Error(event["error"]) |
| 267 | if waitUntil not in self._load_states: |
| 268 | t = deadline - monotonic_time() |
| 269 | if t > 0: |
| 270 | await self._wait_for_load_state_impl(state=waitUntil, timeout=t) |
| 271 | if "newDocument" in event and "request" in event["newDocument"]: |
| 272 | request = from_channel(event["newDocument"]["request"]) |
| 273 | return await request.response() |
| 274 | return None |
| 275 | |
| 276 | return EventContextManagerImpl(asyncio.create_task(continuation())) |
| 277 | |
| 278 | async def wait_for_url( |
| 279 | self, |
no test coverage detected