FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
playwright-python/playwright/_impl/_har_router.py at main · microsoft/playwright-python · GitHub
Uh oh!
There was an error while loading.
Please reload this page
.
microsoft
/
playwright-python
Public
Notifications
You must be signed in to change notification settings
Fork
1.2k
Star
15k
Code
Issues
3
Pull requests
8
Actions
Security and quality
0
Insights
Additional navigation options
Code
Issues
Pull requests
Actions
Security and quality
Insights
Expand file tree
Breadcrumbs
playwright-python
/
playwright
/
_impl
/
_har_router.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
122 lines (109 loc) · 4.39 KB
Breadcrumbs
playwright-python
/
playwright
/
_impl
/
_har_router.py
Copy path
File metadata and controls
122 lines (109 loc) · 4.39 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
:
self
.
_local_utils
.
_channel
.
send_may_fail
(
"harClose"
,
None
, {
"harId"
:
self
.
_har_id
}
)
Back
|
FazBrowse Home
|
New Git URL