(self, route: "Route")
| 58 | ) |
| 59 | |
| 60 | async def _handle(self, route: "Route") -> None: |
| 61 | request = route.request |
| 62 | response: HarLookupResult = await self._local_utils.har_lookup( |
| 63 | harId=self._har_id, |
| 64 | url=request.url, |
| 65 | method=request.method, |
| 66 | headers=await request.headers_array(), |
| 67 | postData=request.post_data_buffer, |
| 68 | isNavigationRequest=request.is_navigation_request(), |
| 69 | ) |
| 70 | action = response["action"] |
| 71 | if action == "redirect": |
| 72 | redirect_url = response["redirectURL"] |
| 73 | assert redirect_url |
| 74 | await route._redirected_navigation_request(redirect_url) |
| 75 | return |
| 76 | |
| 77 | if action == "fulfill": |
| 78 | # If the response status is -1, the request was canceled or stalled, so we just stall it here. |
| 79 | # See https://github.com/microsoft/playwright/issues/29311. |
| 80 | # TODO: it'd be better to abort such requests, but then we likely need to respect the timing, |
| 81 | # because the request might have been stalled for a long time until the very end of the |
| 82 | # test when HAR was recorded but we'd abort it immediately. |
| 83 | if response.get("status") == -1: |
| 84 | return |
| 85 | body = response["body"] |
| 86 | assert body is not None |
| 87 | await route.fulfill( |
| 88 | status=response.get("status"), |
| 89 | headers={ |
| 90 | v["name"]: v["value"] |
| 91 | for v in cast(HeadersArray, response.get("headers", [])) |
| 92 | }, |
| 93 | body=base64.b64decode(body), |
| 94 | ) |
| 95 | return |
| 96 | |
| 97 | if action == "error": |
| 98 | pass |
| 99 | # Report the error, but fall through to the default handler. |
| 100 | |
| 101 | if self._not_found_action == "abort": |
| 102 | await route.abort() |
| 103 | return |
| 104 | |
| 105 | await route.fallback() |
| 106 | |
| 107 | async def add_context_route(self, context: "BrowserContext") -> None: |
| 108 | await context.route( |
no test coverage detected