How to use handle_transport_close 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.

connect_to_browser.py

Source: connect_to_browser.py Github

copy
1import asyncio
2import importlib.metadata
3from typing import Any, Dict, cast
4
5from greenlet import greenlet
6
7from playwright._impl._api_types import Error
8from playwright._impl._browser import Browser as BrowserImpl
9from playwright._impl._connection import Connection, from_channel
10from playwright._impl._object_factory import create_remote_object
11from playwright._impl._playwright import Playwright as PlaywrightImpl
12from playwright._impl._transport import WebSocketTransport
13from playwright.sync_api._generated import Browser as SyncBrowser
14
15
16class ConnectToBrowserContextManager:
17    def __init__(
18        self,
19        ws_endpoint: str,
20        slow_mo: float = None,
21        headers: Dict[str, str] = None,
22    ) -> None:
23        self._browser: SyncBrowser
24        self._ws_endpoint = ws_endpoint
25        self._slow_mo = slow_mo
26        self._headers = headers
27
28    def _make_connection(self, dispatcher_fiber, object_factory, transport, loop) -> Connection:
29        if importlib.metadata.version('playwright') < '1.15.0':
30            return Connection(
31                dispatcher_fiber,
32                create_remote_object,
33                transport)
34        else:
35            return Connection(
36                dispatcher_fiber,
37                create_remote_object,
38                transport,
39                loop)
40
41    def _setup_browser(self, browser: BrowserImpl):
42        version = importlib.metadata.version('playwright')
43        if version < '1.17.0':
44            browser._is_remote = True
45            browser._is_connected_over_websocket = True
46        elif version < '1.19.0':
47            browser._should_close_connection_on_close = True
48        else:
49            browser._should_close_connection_on_close = True
50            browser._local_utils = self._playwright_impl._utils
51
52    def __enter__(self) -> SyncBrowser:
53        loop: asyncio.AbstractEventLoop
54        own_loop = None
55        try:
56            loop = asyncio.get_running_loop()
57        except RuntimeError:
58            loop = asyncio.new_event_loop()
59            own_loop = loop
60        if loop.is_running():
61            raise Error(
62                """It looks like you are using Playwright Sync API inside the asyncio loop.
63Please use the Async API instead."""
64            )
65
66        def greenlet_main() -> None:
67            loop.run_until_complete(self._connection.run_as_sync())
68
69            if own_loop:
70                loop.run_until_complete(loop.shutdown_asyncgens())
71                loop.close()
72
73        dispatcher_fiber = greenlet(greenlet_main)
74        transport = WebSocketTransport(
75            loop,
76            self._ws_endpoint,
77            self._headers,
78            self._slow_mo)
79        self._connection = self._make_connection(
80            dispatcher_fiber,
81            create_remote_object,
82            transport,
83            loop)
84
85        g_self = greenlet.getcurrent()
86
87        def callback_wrapper(playwright_impl: PlaywrightImpl) -> None:
88            self._playwright_impl = playwright_impl
89            g_self.switch()
90
91        self._connection.call_on_object_with_known_name(
92            "Playwright", callback_wrapper)
93
94        dispatcher_fiber.switch()
95
96        pre_launched_browser = self._playwright_impl._initializer.get(
97            "preLaunchedBrowser")
98        assert pre_launched_browser
99        browser = cast(BrowserImpl, from_channel(pre_launched_browser))
100        self._setup_browser(browser)
101
102        def handle_transport_close() -> None:
103            for context in browser.contexts:
104                for page in context.pages:
105                    page._on_close()
106                context._on_close()
107            browser._on_close()
108
109        transport.once("close", handle_transport_close)
110
111        self._browser = SyncBrowser(browser)
112        return self._browser
113
114    def start(self) -> SyncBrowser:
115        return self.__enter__()
116
117    def __exit__(self, *args: Any) -> None:
118        self._browser.close()
119
120
121def connect_to_browser(
122    ws_endpoint: str,
123    slow_mo: float = None,
124    headers: Dict[str, str] = None,
125) -> SyncBrowser:
126    return ConnectToBrowserContextManager(ws_endpoint, slow_mo, headers)
127
Full Screen

_browser_type.py

Source: _browser_type.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 pathlib
17from pathlib import Path
18from typing import TYPE_CHECKING, Dict, List, Optional, Union, cast
19
20from playwright._impl._api_structures import (
21    Geolocation,
22    HttpCredentials,
23    ProxySettings,
24    ViewportSize,
25)
26from playwright._impl._api_types import Error
27from playwright._impl._browser import Browser, normalize_context_params
28from playwright._impl._browser_context import BrowserContext
29from playwright._impl._connection import (
30    ChannelOwner,
31    Connection,
32    from_channel,
33    from_nullable_channel,
34)
35from playwright._impl._helper import (
36    ColorScheme,
37    Env,
38    ForcedColors,
39    ReducedMotion,
40    locals_to_params,
41)
42from playwright._impl._transport import WebSocketTransport
43from playwright._impl._wait_helper import throw_on_timeout
44
45if TYPE_CHECKING:
46    from playwright._impl._playwright import Playwright
47
48
49class BrowserType(ChannelOwner):
50    def __init__(
51        self, parent: ChannelOwner, type: str, guid: str, initializer: Dict
52    ) -> None:
53        super().__init__(parent, type, guid, initializer)
54        _playwright: "Playwright"
55
56    def __repr__(self) -> str:
57        return f"<BrowserType name={self.name} executable_path={self.executable_path}>"
58
59    @property
60    def name(self) -> str:
61        return self._initializer["name"]
62
63    @property
64    def executable_path(self) -> str:
65        return self._initializer["executablePath"]
66
67    async def launch(
68        self,
69        executablePath: Union[str, Path] = None,
70        channel: str = None,
71        args: List[str] = None,
72        ignoreDefaultArgs: Union[bool, List[str]] = None,
73        handleSIGINT: bool = None,
74        handleSIGTERM: bool = None,
75        handleSIGHUP: bool = None,
76        timeout: float = None,
77        env: Env = None,
78        headless: bool = None,
79        devtools: bool = None,
80        proxy: ProxySettings = None,
81        downloadsPath: Union[str, Path] = None,
82        slowMo: float = None,
83        tracesDir: Union[pathlib.Path, str] = None,
84        chromiumSandbox: bool = None,
85        firefoxUserPrefs: Dict[str, Union[str, float, bool]] = None,
86    ) -> Browser:
87        params = locals_to_params(locals())
88        normalize_launch_params(params)
89        browser = cast(
90            Browser, from_channel(await self._channel.send("launch", params))
91        )
92        browser._local_utils = self._playwright._utils
93        return browser
94
95    async def launch_persistent_context(
96        self,
97        userDataDir: Union[str, Path],
98        channel: str = None,
99        executablePath: Union[str, Path] = None,
100        args: List[str] = None,
101        ignoreDefaultArgs: Union[bool, List[str]] = None,
102        handleSIGINT: bool = None,
103        handleSIGTERM: bool = None,
104        handleSIGHUP: bool = None,
105        timeout: float = None,
106        env: Env = None,
107        headless: bool = None,
108        devtools: bool = None,
109        proxy: ProxySettings = None,
110        downloadsPath: Union[str, Path] = None,
111        slowMo: float = None,
112        viewport: ViewportSize = None,
113        screen: ViewportSize = None,
114        noViewport: bool = None,
115        ignoreHTTPSErrors: bool = None,
116        javaScriptEnabled: bool = None,
117        bypassCSP: bool = None,
118        userAgent: str = None,
119        locale: str = None,
120        timezoneId: str = None,
121        geolocation: Geolocation = None,
122        permissions: List[str] = None,
123        extraHTTPHeaders: Dict[str, str] = None,
124        offline: bool = None,
125        httpCredentials: HttpCredentials = None,
126        deviceScaleFactor: float = None,
127        isMobile: bool = None,
128        hasTouch: bool = None,
129        colorScheme: ColorScheme = None,
130        reducedMotion: ReducedMotion = None,
131        forcedColors: ForcedColors = None,
132        acceptDownloads: bool = None,
133        tracesDir: Union[pathlib.Path, str] = None,
134        chromiumSandbox: bool = None,
135        recordHarPath: Union[Path, str] = None,
136        recordHarOmitContent: bool = None,
137        recordVideoDir: Union[Path, str] = None,
138        recordVideoSize: ViewportSize = None,
139        baseURL: str = None,
140        strictSelectors: bool = None,
141    ) -> BrowserContext:
142        userDataDir = str(Path(userDataDir))
143        params = locals_to_params(locals())
144        await normalize_context_params(self._connection._is_sync, params)
145        normalize_launch_params(params)
146        context = cast(
147            BrowserContext,
148            from_channel(await self._channel.send("launchPersistentContext", params)),
149        )
150        context._options = params
151        context.tracing._local_utils = self._playwright._utils
152        return context
153
154    async def connect_over_cdp(
155        self,
156        endpointURL: str,
157        timeout: float = None,
158        slow_mo: float = None,
159        headers: Dict[str, str] = None,
160    ) -> Browser:
161        params = locals_to_params(locals())
162        response = await self._channel.send_return_as_dict("connectOverCDP", params)
163        browser = cast(Browser, from_channel(response["browser"]))
164        browser._local_utils = self._playwright._utils
165
166        default_context = cast(
167            Optional[BrowserContext],
168            from_nullable_channel(response.get("defaultContext")),
169        )
170        if default_context:
171            browser._contexts.append(default_context)
172            default_context._browser = browser
173        return browser
174
175    async def connect(
176        self,
177        ws_endpoint: str,
178        timeout: float = None,
179        slow_mo: float = None,
180        headers: Dict[str, str] = None,
181    ) -> Browser:
182        if timeout is None:
183            timeout = 30000
184
185        headers = {**(headers if headers else {}), "x-playwright-browser": self.name}
186
187        transport = WebSocketTransport(
188            self._connection._loop, ws_endpoint, headers, slow_mo
189        )
190        connection = Connection(
191            self._connection._dispatcher_fiber,
192            self._connection._object_factory,
193            transport,
194            self._connection._loop,
195        )
196        connection.mark_as_remote()
197        connection._is_sync = self._connection._is_sync
198        connection._loop.create_task(connection.run())
199        playwright_future = connection.playwright_future
200
201        timeout_future = throw_on_timeout(timeout, Error("Connection timed out"))
202        done, pending = await asyncio.wait(
203            {transport.on_error_future, playwright_future, timeout_future},
204            return_when=asyncio.FIRST_COMPLETED,
205        )
206        if not playwright_future.done():
207            playwright_future.cancel()
208        if not timeout_future.done():
209            timeout_future.cancel()
210        playwright: "Playwright" = next(iter(done)).result()
211        playwright._set_selectors(self._playwright.selectors)
212        self._connection._child_ws_connections.append(connection)
213        pre_launched_browser = playwright._initializer.get("preLaunchedBrowser")
214        assert pre_launched_browser
215        browser = cast(Browser, from_channel(pre_launched_browser))
216        browser._should_close_connection_on_close = True
217        browser._local_utils = self._playwright._utils
218
219        def handle_transport_close() -> None:
220            for context in browser.contexts:
221                for page in context.pages:
222                    page._on_close()
223                context._on_close()
224            browser._on_close()
225            connection.cleanup()
226
227        transport.once("close", handle_transport_close)
228
229        return browser
230
231
232def normalize_launch_params(params: Dict) -> None:
233    if "env" in params:
234        params["env"] = [
235            {"name": name, "value": str(value)}
236            for [name, value] in params["env"].items()
237        ]
238    if "ignoreDefaultArgs" in params:
239        if params["ignoreDefaultArgs"] is True:
240            params["ignoreAllDefaultArgs"] = True
241            del params["ignoreDefaultArgs"]
242    if "executablePath" in params:
243        params["executablePath"] = str(Path(params["executablePath"]))
244    if "downloadsPath" in params:
245        params["downloadsPath"] = str(Path(params["downloadsPath"]))
246
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)