|
31 | 31 | Optional, |
32 | 32 | Pattern, |
33 | 33 | Set, |
34 | | - Tuple, |
35 | 34 | TypedDict, |
36 | 35 | TypeVar, |
37 | 36 | Union, |
38 | 37 | cast, |
39 | 38 | overload, |
40 | 39 | ) |
41 | | -from urllib.parse import ParseResult, urljoin, urlparse, urlunparse |
42 | 40 |
|
43 | 41 | from playwright._impl._errors import ( |
44 | 42 | Error, |
|
47 | 45 | is_target_closed_error, |
48 | 46 | rewrite_error, |
49 | 47 | ) |
50 | | -from playwright._impl._glob import glob_to_regex_pattern |
51 | 48 | from playwright._impl._greenlets import RouteGreenlet |
52 | 49 | from playwright._impl._internal_structures import NameValue |
53 | 50 | from playwright._impl._str_utils import escape_regex_flags |
| 51 | +from playwright._impl._url_match import URLMatch, url_matches |
54 | 52 |
|
55 | 53 | if TYPE_CHECKING: # pragma: no cover |
56 | 54 | from playwright._impl._internal_structures import HeadersArray |
57 | 55 | from playwright._impl._network import Request, Response, Route, WebSocketRoute |
58 | 56 |
|
59 | | -URLMatch = Union[str, Pattern[str], Callable[[str], bool]] |
60 | 57 | URLMatchRequest = Union[ |
61 | 58 | str, Pattern[str], Callable[["Request"], Union[bool, Awaitable[bool]]] |
62 | 59 | ] |
@@ -153,133 +150,6 @@ class FrameNavigatedEvent(TypedDict): |
153 | 150 | Env = Dict[str, Union[str, float, bool]] |
154 | 151 |
|
155 | 152 |
|
156 | | -def url_matches( |
157 | | - base_url: Optional[str], |
158 | | - url_string: str, |
159 | | - match: Optional[URLMatch], |
160 | | - websocket_url: bool = None, |
161 | | -) -> bool: |
162 | | - if not match: |
163 | | - return True |
164 | | - if isinstance(match, str): |
165 | | - match = re.compile( |
166 | | - resolve_glob_to_regex_pattern(base_url, match, websocket_url) |
167 | | - ) |
168 | | - if isinstance(match, Pattern): |
169 | | - return bool(match.search(url_string)) |
170 | | - return match(url_string) |
171 | | - |
172 | | - |
173 | | -def resolve_glob_to_regex_pattern( |
174 | | - base_url: Optional[str], glob: str, websocket_url: bool = None |
175 | | -) -> str: |
176 | | - if websocket_url: |
177 | | - base_url = to_websocket_base_url(base_url) |
178 | | - glob = resolve_glob_base(base_url, glob) |
179 | | - return glob_to_regex_pattern(glob) |
180 | | - |
181 | | - |
182 | | -def to_websocket_base_url(base_url: Optional[str]) -> Optional[str]: |
183 | | - if base_url is not None and re.match(r"^https?://", base_url): |
184 | | - base_url = re.sub(r"^http", "ws", base_url) |
185 | | - return base_url |
186 | | - |
187 | | - |
188 | | -def resolve_glob_base(base_url: Optional[str], match: str) -> str: |
189 | | - if match[0] == "*": |
190 | | - return match |
191 | | - |
192 | | - token_map: Dict[str, str] = {} |
193 | | - |
194 | | - def map_token(original: str, replacement: str) -> str: |
195 | | - if len(original) == 0: |
196 | | - return "" |
197 | | - token_map[replacement] = original |
198 | | - return replacement |
199 | | - |
200 | | - # Escaped `\\?` behaves the same as `?` in our glob patterns. |
201 | | - match = match.replace(r"\\?", "?") |
202 | | - # Special case about: URLs as they are not relative to base_url |
203 | | - if ( |
204 | | - match.startswith("about:") |
205 | | - or match.startswith("data:") |
206 | | - or match.startswith("chrome:") |
207 | | - or match.startswith("edge:") |
208 | | - or match.startswith("file:") |
209 | | - ): |
210 | | - # about: and data: URLs are not relative to base_url, so we return them as is. |
211 | | - return match |
212 | | - # Glob symbols may be escaped in the URL and some of them such as ? affect resolution, |
213 | | - # so we replace them with safe components first. |
214 | | - processed_parts = [] |
215 | | - for index, token in enumerate(match.split("/")): |
216 | | - if token in (".", "..", ""): |
217 | | - processed_parts.append(token) |
218 | | - continue |
219 | | - # Handle special case of http*://, note that the new schema has to be |
220 | | - # a web schema so that slashes are properly inserted after domain. |
221 | | - if index == 0 and token.endswith(":"): |
222 | | - # Replace any pattern with http: |
223 | | - if "*" in token or "{" in token: |
224 | | - processed_parts.append(map_token(token, "http:")) |
225 | | - else: |
226 | | - # Preserve explicit schema as is as it may affect trailing slashes after domain. |
227 | | - processed_parts.append(token) |
228 | | - continue |
229 | | - question_index = token.find("?") |
230 | | - if question_index == -1: |
231 | | - processed_parts.append(map_token(token, f"$_{index}_$")) |
232 | | - else: |
233 | | - new_prefix = map_token(token[:question_index], f"$_{index}_$") |
234 | | - new_suffix = map_token(token[question_index:], f"?$_{index}_$") |
235 | | - processed_parts.append(new_prefix + new_suffix) |
236 | | - |
237 | | - relative_path = "/".join(processed_parts) |
238 | | - resolved, case_insensitive_part = resolve_base_url(base_url, relative_path) |
239 | | - |
240 | | - for token, original in token_map.items(): |
241 | | - normalize = case_insensitive_part and token in case_insensitive_part |
242 | | - resolved = resolved.replace( |
243 | | - token, original.lower() if normalize else original, 1 |
244 | | - ) |
245 | | - |
246 | | - return resolved |
247 | | - |
248 | | - |
249 | | -def resolve_base_url( |
250 | | - base_url: Optional[str], given_url: str |
251 | | -) -> Tuple[str, Optional[str]]: |
252 | | - try: |
253 | | - url = nodelike_urlparse( |
254 | | - urljoin(base_url if base_url is not None else "", given_url) |
255 | | - ) |
256 | | - resolved = urlunparse(url) |
257 | | - # Schema and domain are case-insensitive. |
258 | | - hostname_port = ( |
259 | | - url.hostname or "" |
260 | | - ) # can't use parsed.netloc because it includes userinfo (username:password) |
261 | | - if url.port: |
262 | | - hostname_port += f":{url.port}" |
263 | | - case_insensitive_prefix = f"{url.scheme}://{hostname_port}" |
264 | | - return resolved, case_insensitive_prefix |
265 | | - except Exception: |
266 | | - return given_url, None |
267 | | - |
268 | | - |
269 | | -def nodelike_urlparse(url: str) -> ParseResult: |
270 | | - parsed = urlparse(url, allow_fragments=True) |
271 | | - |
272 | | - # https://url.spec.whatwg.org/#special-scheme |
273 | | - is_special_url = parsed.scheme in ["http", "https", "ws", "wss", "ftp", "file"] |
274 | | - if is_special_url: |
275 | | - # special urls have a list path, list paths are serialized as follows: https://url.spec.whatwg.org/#url-path-serializer |
276 | | - # urllib diverges, so we patch it here |
277 | | - if parsed.path == "": |
278 | | - parsed = parsed._replace(path="/") |
279 | | - |
280 | | - return parsed |
281 | | - |
282 | | - |
283 | 153 | class HarLookupResult(TypedDict, total=False): |
284 | 154 | action: Literal["error", "redirect", "fulfill", "noentry"] |
285 | 155 | message: Optional[str] |
|
0 commit comments