How to use _disable_interception method in Playwright Python

Best Python code snippet using playwright-python

Run Playwright Python automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

_browser_context.py

Source: _browser_context.py Github

copy
1# Copyright (c) Microsoft Corporation.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import asyncio
16import json
17from pathlib import Path
18from types import SimpleNamespace
19from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Set, Union, cast
20
21from playwright._impl._api_structures import (
22    Cookie,
23    Geolocation,
24    SetCookieParam,
25    StorageState,
26)
27from playwright._impl._api_types import Error
28from playwright._impl._artifact import Artifact
29from playwright._impl._cdp_session import CDPSession
30from playwright._impl._connection import (
31    ChannelOwner,
32    from_channel,
33    from_nullable_channel,
34)
35from playwright._impl._event_context_manager import EventContextManagerImpl
36from playwright._impl._fetch import APIRequestContext
37from playwright._impl._frame import Frame
38from playwright._impl._helper import (
39    RouteHandler,
40    RouteHandlerCallback,
41    TimeoutSettings,
42    URLMatch,
43    URLMatcher,
44    async_readfile,
45    async_writefile,
46    is_safe_close_error,
47    locals_to_params,
48    to_impl,
49)
50from playwright._impl._network import Request, Response, Route, serialize_headers
51from playwright._impl._page import BindingCall, Page, Worker
52from playwright._impl._tracing import Tracing
53from playwright._impl._wait_helper import WaitHelper
54
55if TYPE_CHECKING:  # pragma: no cover
56    from playwright._impl._browser import Browser
57
58
59class BrowserContext(ChannelOwner):
60
61    Events = SimpleNamespace(
62        BackgroundPage="backgroundpage",
63        Close="close",
64        Page="page",
65        ServiceWorker="serviceworker",
66        Request="request",
67        Response="response",
68        RequestFailed="requestfailed",
69        RequestFinished="requestfinished",
70    )
71
72    def __init__(
73        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
74    ) -> None:
75        super().__init__(parent, type, guid, initializer)
76        self._pages: List[Page] = []
77        self._routes: List[RouteHandler] = []
78        self._bindings: Dict[str, Any] = {}
79        self._timeout_settings = TimeoutSettings(None)
80        self._browser: Optional["Browser"] = None
81        self._owner_page: Optional[Page] = None
82        self._options: Dict[str, Any] = {}
83        self._background_pages: Set[Page] = set()
84        self._service_workers: Set[Worker] = set()
85        self._tracing = cast(Tracing, from_channel(initializer["tracing"]))
86        self._request: APIRequestContext = from_channel(
87            initializer["APIRequestContext"]
88        )
89        self._channel.on(
90            "bindingCall",
91            lambda params: self._on_binding(from_channel(params["binding"])),
92        )
93        self._channel.on("close", lambda _: self._on_close())
94        self._channel.on(
95            "page", lambda params: self._on_page(from_channel(params["page"]))
96        )
97        self._channel.on(
98            "route",
99            lambda params: self._on_route(
100                from_channel(params.get("route")), from_channel(params.get("request"))
101            ),
102        )
103
104        self._channel.on(
105            "backgroundPage",
106            lambda params: self._on_background_page(from_channel(params["page"])),
107        )
108
109        self._channel.on(
110            "serviceWorker",
111            lambda params: self._on_service_worker(from_channel(params["worker"])),
112        )
113        self._channel.on(
114            "request",
115            lambda params: self._on_request(
116                from_channel(params["request"]),
117                from_nullable_channel(params.get("page")),
118            ),
119        )
120        self._channel.on(
121            "response",
122            lambda params: self._on_response(
123                from_channel(params["response"]),
124                from_nullable_channel(params.get("page")),
125            ),
126        )
127        self._channel.on(
128            "requestFailed",
129            lambda params: self._on_request_failed(
130                from_channel(params["request"]),
131                params["responseEndTiming"],
132                params.get("failureText"),
133                from_nullable_channel(params.get("page")),
134            ),
135        )
136        self._channel.on(
137            "requestFinished",
138            lambda params: self._on_request_finished(
139                from_channel(params["request"]),
140                from_nullable_channel(params.get("response")),
141                params["responseEndTiming"],
142                from_nullable_channel(params.get("page")),
143            ),
144        )
145        self._closed_future: asyncio.Future = asyncio.Future()
146        self.once(
147            self.Events.Close, lambda context: self._closed_future.set_result(True)
148        )
149
150    def __repr__(self) -> str:
151        return f"<BrowserContext browser={self.browser}>"
152
153    def _on_page(self, page: Page) -> None:
154        self._pages.append(page)
155        self.emit(BrowserContext.Events.Page, page)
156        if page._opener and not page._opener.is_closed():
157            page._opener.emit(Page.Events.Popup, page)
158
159    def _on_route(self, route: Route, request: Request) -> None:
160        for handler_entry in self._routes:
161            if handler_entry.matches(request.url):
162                try:
163                    handler_entry.handle(route, request)
164                finally:
165                    if not handler_entry.is_active:
166                        self._routes.remove(handler_entry)
167                        if not len(self._routes) == 0:
168                            asyncio.create_task(self._disable_interception())
169                break
170        route._internal_continue()
171
172    def _on_binding(self, binding_call: BindingCall) -> None:
173        func = self._bindings.get(binding_call._initializer["name"])
174        if func is None:
175            return
176        asyncio.create_task(binding_call.call(func))
177
178    def set_default_navigation_timeout(self, timeout: float) -> None:
179        self._timeout_settings.set_navigation_timeout(timeout)
180        self._channel.send_no_reply(
181            "setDefaultNavigationTimeoutNoReply", dict(timeout=timeout)
182        )
183
184    def set_default_timeout(self, timeout: float) -> None:
185        self._timeout_settings.set_timeout(timeout)
186        self._channel.send_no_reply("setDefaultTimeoutNoReply", dict(timeout=timeout))
187
188    @property
189    def pages(self) -> List[Page]:
190        return self._pages.copy()
191
192    @property
193    def browser(self) -> Optional["Browser"]:
194        return self._browser
195
196    async def new_page(self) -> Page:
197        if self._owner_page:
198            raise Error("Please use browser.new_context()")
199        return from_channel(await self._channel.send("newPage"))
200
201    async def cookies(self, urls: Union[str, List[str]] = None) -> List[Cookie]:
202        if urls is None:
203            urls = []
204        if not isinstance(urls, list):
205            urls = [urls]
206        return await self._channel.send("cookies", dict(urls=urls))
207
208    async def add_cookies(self, cookies: List[SetCookieParam]) -> None:
209        await self._channel.send("addCookies", dict(cookies=cookies))
210
211    async def clear_cookies(self) -> None:
212        await self._channel.send("clearCookies")
213
214    async def grant_permissions(
215        self, permissions: List[str], origin: str = None
216    ) -> None:
217        await self._channel.send("grantPermissions", locals_to_params(locals()))
218
219    async def clear_permissions(self) -> None:
220        await self._channel.send("clearPermissions")
221
222    async def set_geolocation(self, geolocation: Geolocation = None) -> None:
223        await self._channel.send("setGeolocation", locals_to_params(locals()))
224
225    async def set_extra_http_headers(self, headers: Dict[str, str]) -> None:
226        await self._channel.send(
227            "setExtraHTTPHeaders", dict(headers=serialize_headers(headers))
228        )
229
230    async def set_offline(self, offline: bool) -> None:
231        await self._channel.send("setOffline", dict(offline=offline))
232
233    async def add_init_script(
234        self, script: str = None, path: Union[str, Path] = None
235    ) -> None:
236        if path:
237            script = (await async_readfile(path)).decode()
238        if not isinstance(script, str):
239            raise Error("Either path or script parameter must be specified")
240        await self._channel.send("addInitScript", dict(source=script))
241
242    async def expose_binding(
243        self, name: str, callback: Callable, handle: bool = None
244    ) -> None:
245        for page in self._pages:
246            if name in page._bindings:
247                raise Error(
248                    f'Function "{name}" has been already registered in one of the pages'
249                )
250        if name in self._bindings:
251            raise Error(f'Function "{name}" has been already registered')
252        self._bindings[name] = callback
253        await self._channel.send(
254            "exposeBinding", dict(name=name, needsHandle=handle or False)
255        )
256
257    async def expose_function(self, name: str, callback: Callable) -> None:
258        await self.expose_binding(name, lambda source, *args: callback(*args))
259
260    async def route(
261        self, url: URLMatch, handler: RouteHandlerCallback, times: int = None
262    ) -> None:
263        self._routes.insert(
264            0,
265            RouteHandler(URLMatcher(self._options.get("baseURL"), url), handler, times),
266        )
267        if len(self._routes) == 1:
268            await self._channel.send(
269                "setNetworkInterceptionEnabled", dict(enabled=True)
270            )
271
272    async def unroute(
273        self, url: URLMatch, handler: Optional[RouteHandlerCallback] = None
274    ) -> None:
275        self._routes = list(
276            filter(
277                lambda r: r.matcher.match != url or (handler and r.handler != handler),
278                self._routes,
279            )
280        )
281        if len(self._routes) == 0:
282            await self._disable_interception()
283
284    async def _disable_interception(self) -> None:
285        await self._channel.send("setNetworkInterceptionEnabled", dict(enabled=False))
286
287    def expect_event(
288        self,
289        event: str,
290        predicate: Callable = None,
291        timeout: float = None,
292    ) -> EventContextManagerImpl:
293        if timeout is None:
294            timeout = self._timeout_settings.timeout()
295        wait_helper = WaitHelper(self, f"browser_context.expect_event({event})")
296        wait_helper.reject_on_timeout(
297            timeout, f'Timeout {timeout}ms exceeded while waiting for event "{event}"'
298        )
299        if event != BrowserContext.Events.Close:
300            wait_helper.reject_on_event(
301                self, BrowserContext.Events.Close, Error("Context closed")
302            )
303        wait_helper.wait_for_event(self, event, predicate)
304        return EventContextManagerImpl(wait_helper.result())
305
306    def _on_close(self) -> None:
307        if self._browser:
308            self._browser._contexts.remove(self)
309
310        self.emit(BrowserContext.Events.Close, self)
311
312    async def close(self) -> None:
313        try:
314            if self._options.get("recordHar"):
315                har = cast(
316                    Artifact, from_channel(await self._channel.send("harExport"))
317                )
318                await har.save_as(
319                    cast(Dict[str, str], self._options["recordHar"])["path"]
320                )
321                await har.delete()
322            await self._channel.send("close")
323            await self._closed_future
324        except Exception as e:
325            if not is_safe_close_error(e):
326                raise e
327
328    async def _pause(self) -> None:
329        await self._channel.send("pause")
330
331    async def storage_state(self, path: Union[str, Path] = None) -> StorageState:
332        result = await self._channel.send_return_as_dict("storageState")
333        if path:
334            await async_writefile(path, json.dumps(result))
335        return result
336
337    async def wait_for_event(
338        self, event: str, predicate: Callable = None, timeout: float = None
339    ) -> Any:
340        async with self.expect_event(event, predicate, timeout) as event_info:
341            pass
342        return await event_info
343
344    def expect_page(
345        self,
346        predicate: Callable[[Page], bool] = None,
347        timeout: float = None,
348    ) -> EventContextManagerImpl[Page]:
349        return self.expect_event(BrowserContext.Events.Page, predicate, timeout)
350
351    def _on_background_page(self, page: Page) -> None:
352        self._background_pages.add(page)
353        self.emit(BrowserContext.Events.BackgroundPage, page)
354
355    def _on_service_worker(self, worker: Worker) -> None:
356        worker._context = self
357        self._service_workers.add(worker)
358        self.emit(BrowserContext.Events.ServiceWorker, worker)
359
360    def _on_request_failed(
361        self,
362        request: Request,
363        response_end_timing: float,
364        failure_text: Optional[str],
365        page: Optional[Page],
366    ) -> None:
367        request._failure_text = failure_text
368        if request._timing:
369            request._timing["responseEnd"] = response_end_timing
370        self.emit(BrowserContext.Events.RequestFailed, request)
371        if page:
372            page.emit(Page.Events.RequestFailed, request)
373
374    def _on_request_finished(
375        self,
376        request: Request,
377        response: Optional[Response],
378        response_end_timing: float,
379        page: Optional[Page],
380    ) -> None:
381        if request._timing:
382            request._timing["responseEnd"] = response_end_timing
383        self.emit(BrowserContext.Events.RequestFinished, request)
384        if page:
385            page.emit(Page.Events.RequestFinished, request)
386        if response:
387            response._finished_future.set_result(True)
388
389    def _on_request(self, request: Request, page: Optional[Page]) -> None:
390        self.emit(BrowserContext.Events.Request, request)
391        if page:
392            page.emit(Page.Events.Request, request)
393
394    def _on_response(self, response: Response, page: Optional[Page]) -> None:
395        self.emit(BrowserContext.Events.Response, response)
396        if page:
397            page.emit(Page.Events.Response, response)
398
399    @property
400    def background_pages(self) -> List[Page]:
401        return list(self._background_pages)
402
403    @property
404    def service_workers(self) -> List[Worker]:
405        return list(self._service_workers)
406
407    async def new_cdp_session(self, page: Union[Page, Frame]) -> CDPSession:
408        page = to_impl(page)
409        params = {}
410        if isinstance(page, Page):
411            params["page"] = page._channel
412        elif isinstance(page, Frame):
413            params["frame"] = page._channel
414        else:
415            raise Error("page: expected Page or Frame")
416        return from_channel(await self._channel.send("newCDPSession", params))
417
418    @property
419    def tracing(self) -> Tracing:
420        return self._tracing
421
422    @property
423    def request(self) -> "APIRequestContext":
424        return self._request
425
Full Screen

_page.py

Source: _page.py Github

copy
1# Copyright (c) Microsoft Corporation.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import asyncio
16import base64
17import inspect
18import re
19import sys
20from pathlib import Path
21from types import SimpleNamespace
22from typing import (
23    TYPE_CHECKING,
24    Any,
25    Callable,
26    Dict,
27    List,
28    Optional,
29    Pattern,
30    Union,
31    cast,
32)
33
34from playwright._impl._accessibility import Accessibility
35from playwright._impl._api_structures import (
36    FilePayload,
37    FloatRect,
38    PdfMargins,
39    Position,
40    ViewportSize,
41)
42from playwright._impl._api_types import Error
43from playwright._impl._artifact import Artifact
44from playwright._impl._connection import (
45    ChannelOwner,
46    from_channel,
47    from_nullable_channel,
48)
49from playwright._impl._console_message import ConsoleMessage
50from playwright._impl._dialog import Dialog
51from playwright._impl._download import Download
52from playwright._impl._element_handle import ElementHandle
53from playwright._impl._event_context_manager import EventContextManagerImpl
54from playwright._impl._file_chooser import FileChooser
55from playwright._impl._frame import Frame
56from playwright._impl._helper import (
57    ColorScheme,
58    DocumentLoadState,
59    ForcedColors,
60    KeyboardModifier,
61    MouseButton,
62    ReducedMotion,
63    RouteHandler,
64    RouteHandlerCallback,
65    TimeoutSettings,
66    URLMatch,
67    URLMatcher,
68    URLMatchRequest,
69    URLMatchResponse,
70    async_readfile,
71    async_writefile,
72    is_safe_close_error,
73    locals_to_params,
74    make_dirs_for_file,
75    parse_error,
76    serialize_error,
77)
78from playwright._impl._input import Keyboard, Mouse, Touchscreen
79from playwright._impl._js_handle import (
80    JSHandle,
81    Serializable,
82    parse_result,
83    serialize_argument,
84)
85from playwright._impl._network import Request, Response, Route, serialize_headers
86from playwright._impl._video import Video
87from playwright._impl._wait_helper import WaitHelper
88
89if sys.version_info >= (3, 8):  # pragma: no cover
90    from typing import Literal
91else:  # pragma: no cover
92    from typing_extensions import Literal
93
94if TYPE_CHECKING:  # pragma: no cover
95    from playwright._impl._browser_context import BrowserContext
96    from playwright._impl._fetch import APIRequestContext
97    from playwright._impl._locator import FrameLocator, Locator
98    from playwright._impl._network import WebSocket
99
100
101class Page(ChannelOwner):
102
103    Events = SimpleNamespace(
104        Close="close",
105        Crash="crash",
106        Console="console",
107        Dialog="dialog",
108        Download="download",
109        FileChooser="filechooser",
110        DOMContentLoaded="domcontentloaded",
111        PageError="pageerror",
112        Request="request",
113        Response="response",
114        RequestFailed="requestfailed",
115        RequestFinished="requestfinished",
116        FrameAttached="frameattached",
117        FrameDetached="framedetached",
118        FrameNavigated="framenavigated",
119        Load="load",
120        Popup="popup",
121        WebSocket="websocket",
122        Worker="worker",
123    )
124    accessibility: Accessibility
125    keyboard: Keyboard
126    mouse: Mouse
127    touchscreen: Touchscreen
128
129    def __init__(
130        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
131    ) -> None:
132        super().__init__(parent, type, guid, initializer)
133        self._browser_context: BrowserContext = parent
134        self.accessibility = Accessibility(self._channel)
135        self.keyboard = Keyboard(self._channel)
136        self.mouse = Mouse(self._channel)
137        self.touchscreen = Touchscreen(self._channel)
138
139        self._main_frame: Frame = from_channel(initializer["mainFrame"])
140        self._main_frame._page = self
141        self._frames = [self._main_frame]
142        self._viewport_size: Optional[ViewportSize] = initializer.get("viewportSize")
143        self._is_closed = False
144        self._workers: List["Worker"] = []
145        self._bindings: Dict[str, Any] = {}
146        self._routes: List[RouteHandler] = []
147        self._owned_context: Optional["BrowserContext"] = None
148        self._timeout_settings: TimeoutSettings = TimeoutSettings(
149            self._browser_context._timeout_settings
150        )
151        self._video: Optional[Video] = None
152        self._opener = cast("Page", from_nullable_channel(initializer.get("opener")))
153
154        self._channel.on(
155            "bindingCall",
156            lambda params: self._on_binding(from_channel(params["binding"])),
157        )
158        self._channel.on("close", lambda _: self._on_close())
159        self._channel.on(
160            "console",
161            lambda params: self.emit(
162                Page.Events.Console, from_channel(params["message"])
163            ),
164        )
165        self._channel.on("crash", lambda _: self._on_crash())
166        self._channel.on("dialog", lambda params: self._on_dialog(params))
167        self._channel.on(
168            "domcontentloaded", lambda _: self.emit(Page.Events.DOMContentLoaded, self)
169        )
170        self._channel.on("download", lambda params: self._on_download(params))
171        self._channel.on(
172            "fileChooser",
173            lambda params: self.emit(
174                Page.Events.FileChooser,
175                FileChooser(
176                    self, from_channel(params["element"]), params["isMultiple"]
177                ),
178            ),
179        )
180        self._channel.on(
181            "frameAttached",
182            lambda params: self._on_frame_attached(from_channel(params["frame"])),
183        )
184        self._channel.on(
185            "frameDetached",
186            lambda params: self._on_frame_detached(from_channel(params["frame"])),
187        )
188        self._channel.on("load", lambda _: self.emit(Page.Events.Load, self))
189        self._channel.on(
190            "pageError",
191            lambda params: self.emit(
192                Page.Events.PageError, parse_error(params["error"]["error"])
193            ),
194        )
195        self._channel.on(
196            "route",
197            lambda params: self._on_route(
198                from_channel(params["route"]), from_channel(params["request"])
199            ),
200        )
201        self._channel.on("video", lambda params: self._on_video(params))
202        self._channel.on(
203            "webSocket",
204            lambda params: self.emit(
205                Page.Events.WebSocket, from_channel(params["webSocket"])
206            ),
207        )
208        self._channel.on(
209            "worker", lambda params: self._on_worker(from_channel(params["worker"]))
210        )
211        self._closed_or_crashed_future: asyncio.Future = asyncio.Future()
212        self.on(
213            Page.Events.Close,
214            lambda _: self._closed_or_crashed_future.set_result(True)
215            if not self._closed_or_crashed_future.done()
216            else None,
217        )
218        self.on(
219            Page.Events.Crash,
220            lambda _: self._closed_or_crashed_future.set_result(True)
221            if not self._closed_or_crashed_future.done()
222            else None,
223        )
224
225    def __repr__(self) -> str:
226        return f"<Page url={self.url!r}>"
227
228    def _on_frame_attached(self, frame: Frame) -> None:
229        frame._page = self
230        self._frames.append(frame)
231        self.emit(Page.Events.FrameAttached, frame)
232
233    def _on_frame_detached(self, frame: Frame) -> None:
234        self._frames.remove(frame)
235        frame._detached = True
236        self.emit(Page.Events.FrameDetached, frame)
237
238    def _on_route(self, route: Route, request: Request) -> None:
239        for handler_entry in self._routes:
240            if handler_entry.matches(request.url):
241                try:
242                    handler_entry.handle(route, request)
243                finally:
244                    if not handler_entry.is_active:
245                        self._routes.remove(handler_entry)
246                        if len(self._routes) == 0:
247                            asyncio.create_task(self._disable_interception())
248                return
249        self._browser_context._on_route(route, request)
250
251    def _on_binding(self, binding_call: "BindingCall") -> None:
252        func = self._bindings.get(binding_call._initializer["name"])
253        if func:
254            asyncio.create_task(binding_call.call(func))
255        self._browser_context._on_binding(binding_call)
256
257    def _on_worker(self, worker: "Worker") -> None:
258        self._workers.append(worker)
259        worker._page = self
260        self.emit(Page.Events.Worker, worker)
261
262    def _on_close(self) -> None:
263        self._is_closed = True
264        if self in self._browser_context._pages:
265            self._browser_context._pages.remove(self)
266        if self in self._browser_context._background_pages:
267            self._browser_context._background_pages.remove(self)
268        self.emit(Page.Events.Close, self)
269
270    def _on_crash(self) -> None:
271        self.emit(Page.Events.Crash, self)
272
273    def _on_dialog(self, params: Any) -> None:
274        dialog = cast(Dialog, from_channel(params["dialog"]))
275        if self.listeners(Page.Events.Dialog):
276            self.emit(Page.Events.Dialog, dialog)
277        else:
278            if dialog.type == "beforeunload":
279                asyncio.create_task(dialog.accept())
280            else:
281                asyncio.create_task(dialog.dismiss())
282
283    def _on_download(self, params: Any) -> None:
284        url = params["url"]
285        suggested_filename = params["suggestedFilename"]
286        artifact = cast(Artifact, from_channel(params["artifact"]))
287        self.emit(
288            Page.Events.Download, Download(self, url, suggested_filename, artifact)
289        )
290
291    def _on_video(self, params: Any) -> None:
292        artifact = from_channel(params["artifact"])
293        cast(Video, self.video)._artifact_ready(artifact)
294
295    def _add_event_handler(self, event: str, k: Any, v: Any) -> None:
296        if event == Page.Events.FileChooser and len(self.listeners(event)) == 0:
297            self._channel.send_no_reply(
298                "setFileChooserInterceptedNoReply", {"intercepted": True}
299            )
300        super()._add_event_handler(event, k, v)
301
302    def remove_listener(self, event: str, f: Any) -> None:
303        super().remove_listener(event, f)
304        if event == Page.Events.FileChooser and len(self.listeners(event)) == 0:
305            self._channel.send_no_reply(
306                "setFileChooserInterceptedNoReply", {"intercepted": False}
307            )
308
309    @property
310    def context(self) -> "BrowserContext":
311        return self._browser_context
312
313    async def opener(self) -> Optional["Page"]:
314        if self._opener and self._opener.is_closed():
315            return None
316        return self._opener
317
318    @property
319    def main_frame(self) -> Frame:
320        return self._main_frame
321
322    def frame(self, name: str = None, url: URLMatch = None) -> Optional[Frame]:
323        matcher = (
324            URLMatcher(self._browser_context._options.get("baseURL"), url)
325            if url
326            else None
327        )
328        for frame in self._frames:
329            if name and frame.name == name:
330                return frame
331            if url and matcher and matcher.matches(frame.url):
332                return frame
333        return None
334
335    @property
336    def frames(self) -> List[Frame]:
337        return self._frames.copy()
338
339    def set_default_navigation_timeout(self, timeout: float) -> None:
340        self._timeout_settings.set_navigation_timeout(timeout)
341        self._channel.send_no_reply(
342            "setDefaultNavigationTimeoutNoReply", dict(timeout=timeout)
343        )
344
345    def set_default_timeout(self, timeout: float) -> None:
346        self._timeout_settings.set_timeout(timeout)
347        self._channel.send_no_reply("setDefaultTimeoutNoReply", dict(timeout=timeout))
348
349    async def query_selector(
350        self,
351        selector: str,
352        strict: bool = None,
353    ) -> Optional[ElementHandle]:
354        return await self._main_frame.query_selector(selector, strict)
355
356    async def query_selector_all(self, selector: str) -> List[ElementHandle]:
357        return await self._main_frame.query_selector_all(selector)
358
359    async def wait_for_selector(
360        self,
361        selector: str,
362        timeout: float = None,
363        state: Literal["attached", "detached", "hidden", "visible"] = None,
364        strict: bool = None,
365    ) -> Optional[ElementHandle]:
366        return await self._main_frame.wait_for_selector(**locals_to_params(locals()))
367
368    async def is_checked(
369        self, selector: str, strict: bool = None, timeout: float = None
370    ) -> bool:
371        return await self._main_frame.is_checked(**locals_to_params(locals()))
372
373    async def is_disabled(
374        self, selector: str, strict: bool = None, timeout: float = None
375    ) -> bool:
376        return await self._main_frame.is_disabled(**locals_to_params(locals()))
377
378    async def is_editable(
379        self, selector: str, strict: bool = None, timeout: float = None
380    ) -> bool:
381        return await self._main_frame.is_editable(**locals_to_params(locals()))
382
383    async def is_enabled(
384        self, selector: str, strict: bool = None, timeout: float = None
385    ) -> bool:
386        return await self._main_frame.is_enabled(**locals_to_params(locals()))
387
388    async def is_hidden(
389        self, selector: str, strict: bool = None, timeout: float = None
390    ) -> bool:
391        return await self._main_frame.is_hidden(**locals_to_params(locals()))
392
393    async def is_visible(
394        self, selector: str, strict: bool = None, timeout: float = None
395    ) -> bool:
396        return await self._main_frame.is_visible(**locals_to_params(locals()))
397
398    async def dispatch_event(
399        self,
400        selector: str,
401        type: str,
402        eventInit: Dict = None,
403        timeout: float = None,
404        strict: bool = None,
405    ) -> None:
406        return await self._main_frame.dispatch_event(**locals_to_params(locals()))
407
408    async def evaluate(self, expression: str, arg: Serializable = None) -> Any:
409        return await self._main_frame.evaluate(expression, arg)
410
411    async def evaluate_handle(
412        self, expression: str, arg: Serializable = None
413    ) -> JSHandle:
414        return await self._main_frame.evaluate_handle(expression, arg)
415
416    async def eval_on_selector(
417        self,
418        selector: str,
419        expression: str,
420        arg: Serializable = None,
421        strict: bool = None,
422    ) -> Any:
423        return await self._main_frame.eval_on_selector(
424            selector, expression, arg, strict
425        )
426
427    async def eval_on_selector_all(
428        self,
429        selector: str,
430        expression: str,
431        arg: Serializable = None,
432    ) -> Any:
433        return await self._main_frame.eval_on_selector_all(selector, expression, arg)
434
435    async def add_script_tag(
436        self,
437        url: str = None,
438        path: Union[str, Path] = None,
439        content: str = None,
440        type: str = None,
441    ) -> ElementHandle:
442        return await self._main_frame.add_script_tag(**locals_to_params(locals()))
443
444    async def add_style_tag(
445        self, url: str = None, path: Union[str, Path] = None, content: str = None
446    ) -> ElementHandle:
447        return await self._main_frame.add_style_tag(**locals_to_params(locals()))
448
449    async def expose_function(self, name: str, callback: Callable) -> None:
450        await self.expose_binding(name, lambda source, *args: callback(*args))
451
452    async def expose_binding(
453        self, name: str, callback: Callable, handle: bool = None
454    ) -> None:
455        if name in self._bindings:
456            raise Error(f'Function "{name}" has been already registered')
457        if name in self._browser_context._bindings:
458            raise Error(
459                f'Function "{name}" has been already registered in the browser context'
460            )
461        self._bindings[name] = callback
462        await self._channel.send(
463            "exposeBinding", dict(name=name, needsHandle=handle or False)
464        )
465
466    async def set_extra_http_headers(self, headers: Dict[str, str]) -> None:
467        await self._channel.send(
468            "setExtraHTTPHeaders", dict(headers=serialize_headers(headers))
469        )
470
471    @property
472    def url(self) -> str:
473        return self._main_frame.url
474
475    async def content(self) -> str:
476        return await self._main_frame.content()
477
478    async def set_content(
479        self,
480        html: str,
481        timeout: float = None,
482        waitUntil: DocumentLoadState = None,
483    ) -> None:
484        return await self._main_frame.set_content(**locals_to_params(locals()))
485
486    async def goto(
487        self,
488        url: str,
489        timeout: float = None,
490        waitUntil: DocumentLoadState = None,
491        referer: str = None,
492    ) -> Optional[Response]:
493        return await self._main_frame.goto(**locals_to_params(locals()))
494
495    async def reload(
496        self,
497        timeout: float = None,
498        waitUntil: DocumentLoadState = None,
499    ) -> Optional[Response]:
500        return from_nullable_channel(
501            await self._channel.send("reload", locals_to_params(locals()))
502        )
503
504    async def wait_for_load_state(
505        self,
506        state: Literal["domcontentloaded", "load", "networkidle"] = None,
507        timeout: float = None,
508    ) -> None:
509        return await self._main_frame.wait_for_load_state(**locals_to_params(locals()))
510
511    async def wait_for_url(
512        self,
513        url: URLMatch,
514        wait_until: DocumentLoadState = None,
515        timeout: float = None,
516    ) -> None:
517        return await self._main_frame.wait_for_url(**locals_to_params(locals()))
518
519    async def wait_for_event(
520        self, event: str, predicate: Callable = None, timeout: float = None
521    ) -> Any:
522        async with self.expect_event(event, predicate, timeout) as event_info:
523            pass
524        return await event_info
525
526    async def go_back(
527        self,
528        timeout: float = None,
529        waitUntil: DocumentLoadState = None,
530    ) -> Optional[Response]:
531        return from_nullable_channel(
532            await self._channel.send("goBack", locals_to_params(locals()))
533        )
534
535    async def go_forward(
536        self,
537        timeout: float = None,
538        waitUntil: DocumentLoadState = None,
539    ) -> Optional[Response]:
540        return from_nullable_channel(
541            await self._channel.send("goForward", locals_to_params(locals()))
542        )
543
544    async def emulate_media(
545        self,
546        media: Literal["print", "screen"] = None,
547        colorScheme: ColorScheme = None,
548        reducedMotion: ReducedMotion = None,
549        forcedColors: ForcedColors = None,
550    ) -> None:
551        await self._channel.send("emulateMedia", locals_to_params(locals()))
552
553    async def set_viewport_size(self, viewportSize: ViewportSize) -> None:
554        self._viewport_size = viewportSize
555        await self._channel.send("setViewportSize", locals_to_params(locals()))
556
557    @property
558    def viewport_size(self) -> Optional[ViewportSize]:
559        return self._viewport_size
560
561    async def bring_to_front(self) -> None:
562        await self._channel.send("bringToFront")
563
564    async def add_init_script(
565        self, script: str = None, path: Union[str, Path] = None
566    ) -> None:
567        if path:
568            script = (await async_readfile(path)).decode()
569        if not isinstance(script, str):
570            raise Error("Either path or script parameter must be specified")
571        await self._channel.send("addInitScript", dict(source=script))
572
573    async def route(
574        self, url: URLMatch, handler: RouteHandlerCallback, times: int = None
575    ) -> None:
576        self._routes.insert(
577            0,
578            RouteHandler(
579                URLMatcher(self._browser_context._options.get("baseURL"), url),
580                handler,
581                times,
582            ),
583        )
584        if len(self._routes) == 1:
585            await self._channel.send(
586                "setNetworkInterceptionEnabled", dict(enabled=True)
587            )
588
589    async def unroute(
590        self, url: URLMatch, handler: Optional[RouteHandlerCallback] = None
591    ) -> None:
592        self._routes = list(
593            filter(
594                lambda r: r.matcher.match != url or (handler and r.handler != handler),
595                self._routes,
596            )
597        )
598        if len(self._routes) == 0:
599            await self._disable_interception()
600
601    async def _disable_interception(self) -> None:
602        await self._channel.send("setNetworkInterceptionEnabled", dict(enabled=False))
603
604    async def screenshot(
605        self,
606        timeout: float = None,
607        type: Literal["jpeg", "png"] = None,
608        path: Union[str, Path] = None,
609        quality: int = None,
610        omitBackground: bool = None,
611        fullPage: bool = None,
612        clip: FloatRect = None,
613        animations: Literal["allow", "disabled"] = None,
614        caret: Literal["hide", "initial"] = None,
615        scale: Literal["css", "device"] = None,
616        mask: List["Locator"] = None,
617    ) -> bytes:
618        params = locals_to_params(locals())
619        if "path" in params:
620            del params["path"]
621        if "mask" in params:
622            params["mask"] = list(
623                map(
624                    lambda locator: (
625                        {
626                            "frame": locator._frame._channel,
627                            "selector": locator._selector,
628                        }
629                    ),
630                    params["mask"],
631                )
632            )
633        encoded_binary = await self._channel.send("screenshot", params)
634        decoded_binary = base64.b64decode(encoded_binary)
635        if path:
636            make_dirs_for_file(path)
637            await async_writefile(path, decoded_binary)
638        return decoded_binary
639
640    async def title(self) -> str:
641        return await self._main_frame.title()
642
643    async def close(self, runBeforeUnload: bool = None) -> None:
644        try:
645            await self._channel.send("close", locals_to_params(locals()))
646            if self._owned_context:
647                await self._owned_context.close()
648        except Exception as e:
649            if not is_safe_close_error(e):
650                raise e
651
652    def is_closed(self) -> bool:
653        return self._is_closed
654
655    async def click(
656        self,
657        selector: str,
658        modifiers: List[KeyboardModifier] = None,
659        position: Position = None,
660        delay: float = None,
661        button: MouseButton = None,
662        clickCount: int = None,
663        timeout: float = None,
664        force: bool = None,
665        noWaitAfter: bool = None,
666        trial: bool = None,
667        strict: bool = None,
668    ) -> None:
669        return await self._main_frame.click(**locals_to_params(locals()))
670
671    async def dblclick(
672        self,
673        selector: str,
674        modifiers: List[KeyboardModifier] = None,
675        position: Position = None,
676        delay: float = None,
677        button: MouseButton = None,
678        timeout: float = None,
679        force: bool = None,
680        noWaitAfter: bool = None,
681        strict: bool = None,
682        trial: bool = None,
683    ) -> None:
684        return await self._main_frame.dblclick(**locals_to_params(locals()))
685
686    async def tap(
687        self,
688        selector: str,
689        modifiers: List[KeyboardModifier] = None,
690        position: Position = None,
691        timeout: float = None,
692        force: bool = None,
693        noWaitAfter: bool = None,
694        strict: bool = None,
695        trial: bool = None,
696    ) -> None:
697        return await self._main_frame.tap(**locals_to_params(locals()))
698
699    async def fill(
700        self,
701        selector: str,
702        value: str,
703        timeout: float = None,
704        noWaitAfter: bool = None,
705        strict: bool = None,
706        force: bool = None,
707    ) -> None:
708        return await self._main_frame.fill(**locals_to_params(locals()))
709
710    def locator(
711        self,
712        selector: str,
713        has_text: Union[str, Pattern] = None,
714        has: "Locator" = None,
715    ) -> "Locator":
716        return self._main_frame.locator(selector, has_text=has_text, has=has)
717
718    def frame_locator(self, selector: str) -> "FrameLocator":
719        return self.main_frame.frame_locator(selector)
720
721    async def focus(
722        self, selector: str, strict: bool = None, timeout: float = None
723    ) -> None:
724        return await self._main_frame.focus(**locals_to_params(locals()))
725
726    async def text_content(
727        self, selector: str, strict: bool = None, timeout: float = None
728    ) -> Optional[str]:
729        return await self._main_frame.text_content(**locals_to_params(locals()))
730
731    async def inner_text(
732        self, selector: str, strict: bool = None, timeout: float = None
733    ) -> str:
734        return await self._main_frame.inner_text(**locals_to_params(locals()))
735
736    async def inner_html(
737        self, selector: str, strict: bool = None, timeout: float = None
738    ) -> str:
739        return await self._main_frame.inner_html(**locals_to_params(locals()))
740
741    async def get_attribute(
742        self, selector: str, name: str, strict: bool = None, timeout: float = None
743    ) -> Optional[str]:
744        return await self._main_frame.get_attribute(**locals_to_params(locals()))
745
746    async def hover(
747        self,
748        selector: str,
749        modifiers: List[KeyboardModifier] = None,
750        position: Position = None,
751        timeout: float = None,
752        force: bool = None,
753        strict: bool = None,
754        trial: bool = None,
755    ) -> None:
756        return await self._main_frame.hover(**locals_to_params(locals()))
757
758    async def drag_and_drop(
759        self,
760        source: str,
761        target: str,
762        sourcePosition: Position = None,
763        targetPosition: Position = None,
764        force: bool = None,
765        noWaitAfter: bool = None,
766        timeout: float = None,
767        strict: bool = None,
768        trial: bool = None,
769    ) -> None:
770        return await self._main_frame.drag_and_drop(**locals_to_params(locals()))
771
772    async def select_option(
773        self,
774        selector: str,
775        value: Union[str, List[str]] = None,
776        index: Union[int, List[int]] = None,
777        label: Union[str, List[str]] = None,
778        element: Union["ElementHandle", List["ElementHandle"]] = None,
779        timeout: float = None,
780        noWaitAfter: bool = None,
781        force: bool = None,
782        strict: bool = None,
783    ) -> List[str]:
784        params = locals_to_params(locals())
785        return await self._main_frame.select_option(**params)
786
787    async def input_value(
788        self, selector: str, strict: bool = None, timeout: float = None
789    ) -> str:
790        params = locals_to_params(locals())
791        return await self._main_frame.input_value(**params)
792
793    async def set_input_files(
794        self,
795        selector: str,
796        files: Union[str, Path, FilePayload, List[Union[str, Path]], List[FilePayload]],
797        timeout: float = None,
798        strict: bool = None,
799        noWaitAfter: bool = None,
800    ) -> None:
801        return await self._main_frame.set_input_files(**locals_to_params(locals()))
802
803    async def type(
804        self,
805        selector: str,
806        text: str,
807        delay: float = None,
808        timeout: float = None,
809        noWaitAfter: bool = None,
810        strict: bool = None,
811    ) -> None:
812        return await self._main_frame.type(**locals_to_params(locals()))
813
814    async def press(
815        self,
816        selector: str,
817        key: str,
818        delay: float = None,
819        timeout: float = None,
820        noWaitAfter: bool = None,
821        strict: bool = None,
822    ) -> None:
823        return await self._main_frame.press(**locals_to_params(locals()))
824
825    async def check(
826        self,
827        selector: str,
828        position: Position = None,
829        timeout: float = None,
830        force: bool = None,
831        noWaitAfter: bool = None,
832        strict: bool = None,
833        trial: bool = None,
834    ) -> None:
835        return await self._main_frame.check(**locals_to_params(locals()))
836
837    async def uncheck(
838        self,
839        selector: str,
840        position: Position = None,
841        timeout: float = None,
842        force: bool = None,
843        noWaitAfter: bool = None,
844        strict: bool = None,
845        trial: bool = None,
846    ) -> None:
847        return await self._main_frame.uncheck(**locals_to_params(locals()))
848
849    async def wait_for_timeout(self, timeout: float) -> None:
850        await self._main_frame.wait_for_timeout(timeout)
851
852    async def wait_for_function(
853        self,
854        expression: str,
855        arg: Serializable = None,
856        timeout: float = None,
857        polling: Union[float, Literal["raf"]] = None,
858    ) -> JSHandle:
859        return await self._main_frame.wait_for_function(**locals_to_params(locals()))
860
861    @property
862    def workers(self) -> List["Worker"]:
863        return self._workers.copy()
864
865    @property
866    def request(self) -> "APIRequestContext":
867        return self.context.request
868
869    async def pause(self) -> None:
870        await self._browser_context._pause()
871
872    async def pdf(
873        self,
874        scale: float = None,
875        displayHeaderFooter: bool = None,
876        headerTemplate: str = None,
877        footerTemplate: str = None,
878        printBackground: bool = None,
879        landscape: bool = None,
880        pageRanges: str = None,
881        format: str = None,
882        width: Union[str, float] = None,
883        height: Union[str, float] = None,
884        preferCSSPageSize: bool = None,
885        margin: PdfMargins = None,
886        path: Union[str, Path] = None,
887    ) -> bytes:
888        params = locals_to_params(locals())
889        if "path" in params:
890            del params["path"]
891        encoded_binary = await self._channel.send("pdf", params)
892        decoded_binary = base64.b64decode(encoded_binary)
893        if path:
894            make_dirs_for_file(path)
895            await async_writefile(path, decoded_binary)
896        return decoded_binary
897
898    @property
899    def video(
900        self,
901    ) -> Optional[Video]:
902        if not self._video:
903            self._video = Video(self)
904        return self._video
905
906    def expect_event(
907        self,
908        event: str,
909        predicate: Callable = None,
910        timeout: float = None,
911    ) -> EventContextManagerImpl:
912        return self._expect_event(
913            event, predicate, timeout, f'waiting for event "{event}"'
914        )
915
916    def _expect_event(
917        self,
918        event: str,
919        predicate: Callable = None,
920        timeout: float = None,
921        log_line: str = None,
922    ) -> EventContextManagerImpl:
923        if timeout is None:
924            timeout = self._timeout_settings.timeout()
925        wait_helper = WaitHelper(self, f"page.expect_event({event})")
926        wait_helper.reject_on_timeout(
927            timeout, f'Timeout {timeout}ms exceeded while waiting for event "{event}"'
928        )
929        if log_line:
930            wait_helper.log(log_line)
931        if event != Page.Events.Crash:
932            wait_helper.reject_on_event(self, Page.Events.Crash, Error("Page crashed"))
933        if event != Page.Events.Close:
934            wait_helper.reject_on_event(self, Page.Events.Close, Error("Page closed"))
935        wait_helper.wait_for_event(self, event, predicate)
936        return EventContextManagerImpl(wait_helper.result())
937
938    def expect_console_message(
939        self,
940        predicate: Callable[[ConsoleMessage], bool] = None,
941        timeout: float = None,
942    ) -> EventContextManagerImpl[ConsoleMessage]:
943        return self.expect_event(Page.Events.Console, predicate, timeout)
944
945    def expect_download(
946        self,
947        predicate: Callable[[Download], bool] = None,
948        timeout: float = None,
949    ) -> EventContextManagerImpl[Download]:
950        return self.expect_event(Page.Events.Download, predicate, timeout)
951
952    def expect_file_chooser(
953        self,
954        predicate: Callable[[FileChooser], bool] = None,
955        timeout: float = None,
956    ) -> EventContextManagerImpl[FileChooser]:
957        return self.expect_event(Page.Events.FileChooser, predicate, timeout)
958
959    def expect_navigation(
960        self,
961        url: URLMatch = None,
962        wait_until: DocumentLoadState = None,
963        timeout: float = None,
964    ) -> EventContextManagerImpl[Response]:
965        return self.main_frame.expect_navigation(url, wait_until, timeout)
966
967    def expect_popup(
968        self,
969        predicate: Callable[["Page"], bool] = None,
970        timeout: float = None,
971    ) -> EventContextManagerImpl["Page"]:
972        return self.expect_event(Page.Events.Popup, predicate, timeout)
973
974    def expect_request(
975        self,
976        url_or_predicate: URLMatchRequest,
977        timeout: float = None,
978    ) -> EventContextManagerImpl[Request]:
979        matcher = (
980            None
981            if callable(url_or_predicate)
982            else URLMatcher(
983                self._browser_context._options.get("baseURL"), url_or_predicate
984            )
985        )
986        predicate = url_or_predicate if callable(url_or_predicate) else None
987
988        def my_predicate(request: Request) -> bool:
989            if matcher:
990                return matcher.matches(request.url)
991            if predicate:
992                return predicate(request)
993            return True
994
995        trimmed_url = trim_url(url_or_predicate)
996        log_line = f"waiting for request {trimmed_url}" if trimmed_url else None
997        return self._expect_event(
998            Page.Events.Request,
999            predicate=my_predicate,
1000            timeout=timeout,
1001            log_line=log_line,
1002        )
1003
1004    def expect_request_finished(
1005        self,
1006        predicate: Callable[["Request"], bool] = None,
1007        timeout: float = None,
1008    ) -> EventContextManagerImpl[Request]:
1009        return self.expect_event(
1010            Page.Events.RequestFinished, predicate=predicate, timeout=timeout
1011        )
1012
1013    def expect_response(
1014        self,
1015        url_or_predicate: URLMatchResponse,
1016        timeout: float = None,
1017    ) -> EventContextManagerImpl[Response]:
1018        matcher = (
1019            None
1020            if callable(url_or_predicate)
1021            else URLMatcher(
1022                self._browser_context._options.get("baseURL"), url_or_predicate
1023            )
1024        )
1025        predicate = url_or_predicate if callable(url_or_predicate) else None
1026
1027        def my_predicate(response: Response) -> bool:
1028            if matcher:
1029                return matcher.matches(response.url)
1030            if predicate:
1031                return predicate(response)
1032            return True
1033
1034        trimmed_url = trim_url(url_or_predicate)
1035        log_line = f"waiting for response {trimmed_url}" if trimmed_url else None
1036        return self._expect_event(
1037            Page.Events.Response,
1038            predicate=my_predicate,
1039            timeout=timeout,
1040            log_line=log_line,
1041        )
1042
1043    def expect_websocket(
1044        self,
1045        predicate: Callable[["WebSocket"], bool] = None,
1046        timeout: float = None,
1047    ) -> EventContextManagerImpl["WebSocket"]:
1048        return self.expect_event("websocket", predicate, timeout)
1049
1050    def expect_worker(
1051        self,
1052        predicate: Callable[["Worker"], bool] = None,
1053        timeout: float = None,
1054    ) -> EventContextManagerImpl["Worker"]:
1055        return self.expect_event("worker", predicate, timeout)
1056
1057    async def set_checked(
1058        self,
1059        selector: str,
1060        checked: bool,
1061        position: Position = None,
1062        timeout: float = None,
1063        force: bool = None,
1064        noWaitAfter: bool = None,
1065        strict: bool = None,
1066        trial: bool = None,
1067    ) -> None:
1068        if checked:
1069            await self.check(
1070                selector=selector,
1071                position=position,
1072                timeout=timeout,
1073                force=force,
1074                noWaitAfter=noWaitAfter,
1075                strict=strict,
1076                trial=trial,
1077            )
1078        else:
1079            await self.uncheck(
1080                selector=selector,
1081                position=position,
1082                timeout=timeout,
1083                force=force,
1084                noWaitAfter=noWaitAfter,
1085                strict=strict,
1086                trial=trial,
1087            )
1088
1089
1090class Worker(ChannelOwner):
1091    Events = SimpleNamespace(Close="close")
1092
1093    def __init__(
1094        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
1095    ) -> None:
1096        super().__init__(parent, type, guid, initializer)
1097        self._channel.on("close", lambda _: self._on_close())
1098        self._page: Optional[Page] = None
1099        self._context: Optional["BrowserContext"] = None
1100
1101    def __repr__(self) -> str:
1102        return f"<Worker url={self.url!r}>"
1103
1104    def _on_close(self) -> None:
1105        if self._page:
1106            self._page._workers.remove(self)
1107        if self._context:
1108            self._context._service_workers.remove(self)
1109        self.emit(Worker.Events.Close, self)
1110
1111    @property
1112    def url(self) -> str:
1113        return self._initializer["url"]
1114
1115    async def evaluate(self, expression: str, arg: Serializable = None) -> Any:
1116        return parse_result(
1117            await self._channel.send(
1118                "evaluateExpression",
1119                dict(
1120                    expression=expression,
1121                    arg=serialize_argument(arg),
1122                ),
1123            )
1124        )
1125
1126    async def evaluate_handle(
1127        self, expression: str, arg: Serializable = None
1128    ) -> JSHandle:
1129        return from_channel(
1130            await self._channel.send(
1131                "evaluateExpressionHandle",
1132                dict(
1133                    expression=expression,
1134                    arg=serialize_argument(arg),
1135                ),
1136            )
1137        )
1138
1139
1140class BindingCall(ChannelOwner):
1141    def __init__(
1142        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
1143    ) -> None:
1144        super().__init__(parent, type, guid, initializer)
1145
1146    async def call(self, func: Callable) -> None:
1147        try:
1148            frame = from_channel(self._initializer["frame"])
1149            source = dict(context=frame._page.context, page=frame._page, frame=frame)
1150            if self._initializer.get("handle"):
1151                result = func(source, from_channel(self._initializer["handle"]))
1152            else:
1153                func_args = list(map(parse_result, self._initializer["args"]))
1154                result = func(source, *func_args)
1155            if inspect.iscoroutine(result):
1156                result = await result
1157            await self._channel.send("resolve", dict(result=serialize_argument(result)))
1158        except Exception as e:
1159            tb = sys.exc_info()[2]
1160            asyncio.create_task(
1161                self._channel.send(
1162                    "reject", dict(error=dict(error=serialize_error(e, tb)))
1163                )
1164            )
1165
1166
1167def trim_url(param: URLMatchRequest) -> Optional[str]:
1168    if isinstance(param, re.Pattern):
1169        return trim_end(param.pattern)
1170    if isinstance(param, str):
1171        return trim_end(param)
1172    return None
1173
1174
1175def trim_end(s: str) -> str:
1176    if len(s) > 50:
1177        return s[:50] + "\u2026"
1178    return s
1179
Full Screen

Accelerate Your Automation Test Cycles With LambdaTest

Leverage LambdaTest’s cloud-based platform to execute your automation tests in parallel and trim down your test execution time significantly. Your first 100 automation testing minutes are on us.

Try LambdaTest

Run Python Tests on LambdaTest Cloud Grid

Execute automation tests with Playwright Python on a cloud-based Grid of 3000+ real browsers and operating systems for both web and mobile applications.

Test now for Free
LambdaTestX

We use cookies to give you the best experience. Cookies help to provide a more personalized experience and relevant advertising for you, and web analytics for us. Learn More in our Cookies policy, Privacy & Terms of service

Allow Cookie
Sarah

I hope you find the best code examples for your project.

If you want to accelerate automated browser testing, try LambdaTest. Your first 100 automation testing minutes are FREE.

Sarah Elson (Product & Growth Lead)