How to use test_connect_to_an_existing_cdp_session_twice method in Playwright Python

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
15from typing import Dict
16
17import pytest
18import requests
19
20from playwright.async_api import BrowserType
21from tests.server import Server, find_free_port
22
23pytestmark = pytest.mark.only_browser("chromium")
24
25
26async def test_connect_to_an_existing_cdp_session(
27    launch_arguments: Dict, browser_type: BrowserType
28):
29    port = find_free_port()
30    browser_server = await browser_type.launch(
31        **launch_arguments, args=[f"--remote-debugging-port={port}"]
32    )
33    cdp_browser = await browser_type.connect_over_cdp(f"http://localhost:{port}")
34    assert len(cdp_browser.contexts) == 1
35    await cdp_browser.close()
36    await browser_server.close()
37
38
39async def test_connect_to_an_existing_cdp_session_twice(
40    launch_arguments: Dict, browser_type: BrowserType, server: Server
41):
42    port = find_free_port()
43    browser_server = await browser_type.launch(
44        **launch_arguments, args=[f"--remote-debugging-port={port}"]
45    )
46    endpoint_url = f"http://localhost:{port}"
47    cdp_browser1 = await browser_type.connect_over_cdp(endpoint_url)
48    cdp_browser2 = await browser_type.connect_over_cdp(endpoint_url)
49    assert len(cdp_browser1.contexts) == 1
50    page1 = await cdp_browser1.contexts[0].new_page()
51    await page1.goto(server.EMPTY_PAGE)
52
53    assert len(cdp_browser2.contexts) == 1
54    page2 = await cdp_browser2.contexts[0].new_page()
55    await page2.goto(server.EMPTY_PAGE)
56
57    assert len(cdp_browser1.contexts[0].pages) == 2
58    assert len(cdp_browser2.contexts[0].pages) == 2
59
60    await cdp_browser1.close()
61    await cdp_browser2.close()
62    await browser_server.close()
63
64
65def _ws_endpoint_from_url(url: str) -> str:
66    response = requests.get(url)
67    assert response.ok
68    response_body = response.json()
69    return response_body["webSocketDebuggerUrl"]
70
71
72async def test_conect_over_a_ws_endpoint(
73    launch_arguments: Dict, browser_type: BrowserType, server: Server
74):
75    port = find_free_port()
76    browser_server = await browser_type.launch(
77        **launch_arguments, args=[f"--remote-debugging-port={port}"]
78    )
79    ws_endpoint = _ws_endpoint_from_url(f"http://localhost:{port}/json/version/")
80
81    cdp_browser1 = await browser_type.connect_over_cdp(ws_endpoint)
82    assert len(cdp_browser1.contexts) == 1
83    await cdp_browser1.close()
84
85    cdp_browser2 = await browser_type.connect_over_cdp(ws_endpoint)
86    assert len(cdp_browser2.contexts) == 1
87    await cdp_browser2.close()
88    await browser_server.close()
89

Source: test_browsertype_connect_cdp.py

# 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.

from typing import Dict

import pytest
import requests

from playwright.async_api import BrowserType
from tests.server import Server, find_free_port

pytestmark = pytest.mark.only_browser("chromium")


async def test_connect_to_an_existing_cdp_session(
    launch_arguments: Dict, browser_type: BrowserType
):
    port = find_free_port()
    browser_server = await browser_type.launch(
        **launch_arguments, args=[f"--remote-debugging-port={port}"]
    )
    cdp_browser = await browser_type.connect_over_cdp(f"http://localhost:{port}")
    assert len(cdp_browser.contexts) == 1
    await cdp_browser.close()
    await browser_server.close()


async def test_connect_to_an_existing_cdp_session_twice(
    launch_arguments: Dict, browser_type: BrowserType, server: Server
):
    port = find_free_port()
    browser_server = await browser_type.launch(
        **launch_arguments, args=[f"--remote-debugging-port={port}"]
    )
    endpoint_url = f"http://localhost:{port}"
    cdp_browser1 = await browser_type.connect_over_cdp(endpoint_url)
    cdp_browser2 = await browser_type.connect_over_cdp(endpoint_url)
    assert len(cdp_browser1.contexts) == 1
    page1 = await cdp_browser1.contexts[0].new_page()
    await page1.goto(server.EMPTY_PAGE)

    assert len(cdp_browser2.contexts) == 1
    page2 = await cdp_browser2.contexts[0].new_page()
    await page2.goto(server.EMPTY_PAGE)

    assert len(cdp_browser1.contexts[0].pages) == 2
    assert len(cdp_browser2.contexts[0].pages) == 2

    await cdp_browser1.close()
    await cdp_browser2.close()
    await browser_server.close()


def _ws_endpoint_from_url(url: str) -> str:
    response = requests.get(url)
    assert response.ok
    response_body = response.json()
    return response_body["webSocketDebuggerUrl"]


async def test_conect_over_a_ws_endpoint(
    launch_arguments: Dict, browser_type: BrowserType, server: Server
):
    port = find_free_port()
    browser_server = await browser_type.launch(
        **launch_arguments, args=[f"--remote-debugging-port={port}"]
    )
    ws_endpoint = _ws_endpoint_from_url(f"http://localhost:{port}/json/version/")

    cdp_browser1 = await browser_type.connect_over_cdp(ws_endpoint)
    assert len(cdp_browser1.contexts) == 1
    await cdp_browser1.close()

    cdp_browser2 = await browser_type.connect_over_cdp(ws_endpoint)
    assert len(cdp_browser2.contexts) == 1
    await cdp_browser2.close()
    await browser_server.close()