1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
# Copyright (c) Microsoft Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import asyncio
import base64
from typing import TYPE_CHECKING, Optional, cast
from playwright._impl._api_structures import HeadersArray
from playwright._impl._helper import (
HarLookupResult,
RouteFromHarNotFoundPolicy,
URLMatch,
)
from playwright._impl._local_utils import LocalUtils
if TYPE_CHECKING: # pragma: no cover
from playwright._impl._browser_context import BrowserContext
from playwright._impl._network import Route
from playwright._impl._page import Page
class HarRouter:
def __init__(
self,
local_utils: LocalUtils,
har_id: str,
not_found_action: RouteFromHarNotFoundPolicy,
url_matcher: Optional[URLMatch] = None,
) -> None:
self._local_utils: LocalUtils = local_utils
self._har_id: str = har_id
self._not_found_action: RouteFromHarNotFoundPolicy = not_found_action
self._options_url_match: Optional[URLMatch] = url_matcher
@staticmethod
async def create(
local_utils: LocalUtils,
file: str,
not_found_action: RouteFromHarNotFoundPolicy,
url_matcher: Optional[URLMatch] = None,
) -> "HarRouter":
har_id = await local_utils._channel.send("harOpen", None, {"file": file})
return HarRouter(
local_utils=local_utils,
har_id=har_id,
not_found_action=not_found_action,
url_matcher=url_matcher,
)
async def _handle(self, route: "Route") -> None:
request = route.request
response: HarLookupResult = await self._local_utils.har_lookup(
harId=self._har_id,
url=request.url,
method=request.method,
headers=await request.headers_array(),
postData=request.post_data_buffer,
isNavigationRequest=request.is_navigation_request(),
)
action = response["action"]
if action == "redirect":
redirect_url = response["redirectURL"]
assert redirect_url
await route._redirected_navigation_request(redirect_url)
return
if action == "fulfill":
# If the response status is -1, the request was canceled or stalled, so we just stall it here.
# See https://github.com/microsoft/playwright/issues/29311.
# TODO: it'd be better to abort such requests, but then we likely need to respect the timing,
# because the request might have been stalled for a long time until the very end of the
# test when HAR was recorded but we'd abort it immediately.
if response.get("status") == -1:
return
body = response["body"]
assert body is not None
await route.fulfill(
status=response.get("status"),
headers={
v["name"]: v["value"]
for v in cast(HeadersArray, response.get("headers", []))
},
body=base64.b64decode(body),
)
return
if action == "error":
pass
# Report the error, but fall through to the default handler.
if self._not_found_action == "abort":
await route.abort()
return
await route.fallback()
async def add_context_route(self, context: "BrowserContext") -> None:
await context.route(
url=self._options_url_match or "**/*",
handler=lambda route, _: asyncio.create_task(self._handle(route)),
)
async def add_page_route(self, page: "Page") -> None:
await page.route(
url=self._options_url_match or "**/*",
handler=lambda route, _: asyncio.create_task(self._handle(route)),
)
def dispose(self) -> None:
asyncio.create_task(
self._local_utils._channel.send("harClose", None, {"harId": self._har_id})
)
|
