summaryrefslogtreecommitdiffstats
path: root/testing/web-platform/mozilla/tests/webdriver
diff options
context:
space:
mode:
Diffstat (limited to 'testing/web-platform/mozilla/tests/webdriver')
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/__init__.py20
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/reference_context.py72
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/type_hint.py31
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/error.py48
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/errors/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/errors/errors.py8
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/interface/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/interface/interface.py26
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/script/exception_details.py69
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/bidi/websocket_upgrade.py156
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/cdp/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/cdp/debugger_address.py45
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/cdp/port_file.py30
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/scroll_into_view.py50
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/send_keys.py24
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/execute_async.py59
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/chrome.py25
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/chrome.py43
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/bidi_disabled.py33
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/binary.py33
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/conftest.py58
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/create.py11
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/invalid.py53
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/new_session/profile_root.py38
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/protocol/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_hosts.py53
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_origins.py56
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/protocol/marionette_port.py41
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/protocol/request.py72
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/send_alert_text.py22
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/__init__.py12
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/iframe.py47
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/screenshot.py51
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/conftest.py15
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/support/__init__.py0
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/support/context.py20
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/support/fixtures.py256
-rw-r--r--testing/web-platform/mozilla/tests/webdriver/support/network.py78
46 files changed, 1655 insertions, 0 deletions
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/__init__.py
new file mode 100644
index 0000000000..910b202075
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/__init__.py
@@ -0,0 +1,20 @@
+import contextlib
+
+
+def set_context(session, context):
+ session.send_session_command("POST", "moz/context", {"context": context})
+
+
+@contextlib.contextmanager
+def using_context(session, context):
+ orig_context = session.send_session_command("GET", "moz/context")
+ needs_change = context != orig_context
+
+ if needs_change:
+ set_context(session, context)
+
+ try:
+ yield
+ finally:
+ if needs_change:
+ set_context(session, orig_context)
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/reference_context.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/reference_context.py
new file mode 100644
index 0000000000..1a5906339b
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/reference_context.py
@@ -0,0 +1,72 @@
+import pytest
+
+from . import using_context
+
+pytestmark = pytest.mark.asyncio
+
+
+# Helper to assert the order of top level browsing contexts.
+# The window used for the assertion is inferred from the first context id of
+# expected_context_ids.
+def assert_tab_order(session, expected_context_ids):
+ with using_context(session, "chrome"):
+ context_ids = session.execute_script(
+ """
+ const contextId = arguments[0];
+ const { TabManager } =
+ ChromeUtils.importESModule("chrome://remote/content/shared/TabManager.sys.mjs");
+ const browsingContext = TabManager.getBrowsingContextById(contextId);
+ const chromeWindow = browsingContext.embedderElement.ownerGlobal;
+ const tabBrowser = TabManager.getTabBrowser(chromeWindow);
+ return tabBrowser.browsers.map(browser => TabManager.getIdForBrowser(browser));
+ """,
+ args=(expected_context_ids[0],),
+ )
+
+ assert context_ids == expected_context_ids
+
+
+async def test_reference_context(bidi_session, current_session):
+ # Create a new window with a tab tab1
+ result = await bidi_session.browsing_context.create(type_hint="window")
+ tab1_context_id = result["context"]
+
+ # Create a second window with a tab tab2
+ result = await bidi_session.browsing_context.create(type_hint="window")
+ tab2_context_id = result["context"]
+
+ # Create a new tab tab3 next to tab1
+ result = await bidi_session.browsing_context.create(
+ type_hint="tab", reference_context=tab1_context_id
+ )
+ tab3_context_id = result["context"]
+
+ # Create a new tab tab4 next to tab2
+ result = await bidi_session.browsing_context.create(
+ type_hint="tab", reference_context=tab2_context_id
+ )
+ tab4_context_id = result["context"]
+
+ # Create a new tab tab5 also next to tab2 (should consequently be between
+ # tab2 and tab4)
+ result = await bidi_session.browsing_context.create(
+ type_hint="tab", reference_context=tab2_context_id
+ )
+ tab5_context_id = result["context"]
+
+ # Create a new window, but pass a reference_context from an existing window.
+ # The reference context is expected to be ignored here.
+ result = await bidi_session.browsing_context.create(
+ type_hint="window", reference_context=tab2_context_id
+ )
+ tab6_context_id = result["context"]
+
+ # We expect 3 windows in total, with a specific tab order:
+ # - the first window should contain tab1, tab3
+ assert_tab_order(current_session, [tab1_context_id, tab3_context_id])
+ # - the second window should contain tab2, tab5, tab4
+ assert_tab_order(
+ current_session, [tab2_context_id, tab5_context_id, tab4_context_id]
+ )
+ # - the third window should contain tab6
+ assert_tab_order(current_session, [tab6_context_id])
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/type_hint.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/type_hint.py
new file mode 100644
index 0000000000..337a03b3dd
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/create/type_hint.py
@@ -0,0 +1,31 @@
+import pytest
+from tests.support.asserts import assert_success
+
+from . import using_context
+
+pytestmark = pytest.mark.asyncio
+
+
+def count_window_handles(session):
+ with using_context(session, "chrome"):
+ response = session.transport.send(
+ "GET", "session/{session_id}/window/handles".format(**vars(session))
+ )
+ chrome_handles = assert_success(response)
+ return len(chrome_handles)
+
+
+@pytest.mark.parametrize("type_hint", ["tab", "window"])
+async def test_type_hint(bidi_session, current_session, type_hint):
+ assert len(await bidi_session.browsing_context.get_tree()) == 1
+ assert count_window_handles(current_session) == 1
+
+ await bidi_session.browsing_context.create(type_hint=type_hint)
+
+ if type_hint == "window":
+ expected_window_count = 2
+ else:
+ expected_window_count = 1
+
+ assert len(await bidi_session.browsing_context.get_tree()) == 2
+ assert count_window_handles(current_session) == expected_window_count
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/error.py b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/error.py
new file mode 100644
index 0000000000..374359d1ae
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/browsing_context/navigate/error.py
@@ -0,0 +1,48 @@
+import os
+from copy import deepcopy
+
+import pytest
+from tests.bidi.browsing_context.navigate import navigate_and_assert
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_insecure_certificate(configuration, url, custom_profile, geckodriver):
+ try:
+ # Create a new profile and remove the certificate storage so that
+ # loading a HTTPS page will cause an insecure certificate error
+ os.remove(os.path.join(custom_profile.profile, "cert9.db"))
+ except Exception:
+ pass
+
+ config = deepcopy(configuration)
+ config["capabilities"]["moz:firefoxOptions"]["args"] = [
+ "--profile",
+ custom_profile.profile,
+ ]
+ # Capability matching not implemented yet for WebDriver BiDi (bug 1713784)
+ config["capabilities"]["acceptInsecureCerts"] = False
+ config["capabilities"]["webSocketUrl"] = True
+
+ driver = geckodriver(config=config)
+ driver.new_session()
+
+ bidi_session = driver.session.bidi_session
+ await bidi_session.start()
+
+ contexts = await bidi_session.browsing_context.get_tree(max_depth=0)
+ await navigate_and_assert(
+ bidi_session,
+ contexts[0],
+ url("/common/blank.html", protocol="https"),
+ expected_error=True,
+ )
+
+
+async def test_invalid_content_encoding(bidi_session, new_tab, inline):
+ await navigate_and_assert(
+ bidi_session,
+ new_tab,
+ f"{inline('<div>foo')}&pipe=header(Content-Encoding,gzip)",
+ expected_error=True,
+ )
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/errors/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/errors/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/errors/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/errors/errors.py b/testing/web-platform/mozilla/tests/webdriver/bidi/errors/errors.py
new file mode 100644
index 0000000000..69b1f2fb7a
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/errors/errors.py
@@ -0,0 +1,8 @@
+import pytest
+from webdriver.bidi.error import UnknownCommandException
+
+
+@pytest.mark.asyncio
+async def test_internal_method(bidi_session, send_blocking_command):
+ with pytest.raises(UnknownCommandException):
+ await send_blocking_command("log._applySessionData", {})
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/interface/__init__.py b/testing/web-platform/mozilla/tests/webdriver/bidi/interface/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/interface/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/interface/interface.py b/testing/web-platform/mozilla/tests/webdriver/bidi/interface/interface.py
new file mode 100644
index 0000000000..561b80d120
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/interface/interface.py
@@ -0,0 +1,26 @@
+import pytest
+from webdriver.bidi.client import BidiSession
+from webdriver.bidi.modules.script import ContextTarget
+
+pytestmark = pytest.mark.asyncio
+
+
+async def test_navigator_webdriver_enabled(inline, browser):
+ # Request a new browser with only WebDriver BiDi and not Marionette/CDP enabled.
+ current_browser = browser(use_bidi=True, extra_prefs={"remote.active-protocols": 1})
+ server_host = current_browser.remote_agent_host
+ server_port = current_browser.remote_agent_port
+
+ async with BidiSession.bidi_only(
+ f"ws://{server_host}:{server_port}"
+ ) as bidi_session:
+ contexts = await bidi_session.browsing_context.get_tree(max_depth=0)
+ assert len(contexts) > 0
+
+ result = await bidi_session.script.evaluate(
+ expression="navigator.webdriver",
+ target=ContextTarget(contexts[0]["context"]),
+ await_promise=False,
+ )
+
+ assert result == {"type": "boolean", "value": True}
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/script/exception_details.py b/testing/web-platform/mozilla/tests/webdriver/bidi/script/exception_details.py
new file mode 100644
index 0000000000..43bccdb845
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/script/exception_details.py
@@ -0,0 +1,69 @@
+import pytest
+from webdriver.bidi.modules.script import ContextTarget, ScriptEvaluateResultException
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("await_promise", [True, False])
+@pytest.mark.parametrize(
+ "expression",
+ [
+ "null",
+ "{ toString: 'not a function' }",
+ "{ toString: () => {{ throw 'toString not allowed'; }} }",
+ "{ toString: () => true }",
+ ],
+)
+@pytest.mark.asyncio
+async def test_call_function_without_to_string_interface(
+ bidi_session, top_context, await_promise, expression
+):
+ function_declaration = "()=>{throw { toString: 'not a function' } }"
+ if await_promise:
+ function_declaration = "async" + function_declaration
+
+ with pytest.raises(ScriptEvaluateResultException) as exception:
+ await bidi_session.script.call_function(
+ function_declaration=function_declaration,
+ await_promise=await_promise,
+ target=ContextTarget(top_context["context"]),
+ )
+
+ assert "exceptionDetails" in exception.value.result
+ exceptionDetails = exception.value.result["exceptionDetails"]
+
+ assert "text" in exceptionDetails
+ assert isinstance(exceptionDetails["text"], str)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("await_promise", [True, False])
+@pytest.mark.parametrize(
+ "expression",
+ [
+ "null",
+ "{ toString: 'not a function' }",
+ "{ toString: () => {{ throw 'toString not allowed'; }} }",
+ "{ toString: () => true }",
+ ],
+)
+@pytest.mark.asyncio
+async def test_evaluate_without_to_string_interface(
+ bidi_session, top_context, await_promise, expression
+):
+ if await_promise:
+ expression = f"Promise.reject({expression})"
+ else:
+ expression = f"throw {expression}"
+
+ with pytest.raises(ScriptEvaluateResultException) as exception:
+ await bidi_session.script.evaluate(
+ expression=expression,
+ await_promise=await_promise,
+ target=ContextTarget(top_context["context"]),
+ )
+
+ assert "exceptionDetails" in exception.value.result
+ exceptionDetails = exception.value.result["exceptionDetails"]
+
+ assert "text" in exceptionDetails
+ assert isinstance(exceptionDetails["text"], str)
diff --git a/testing/web-platform/mozilla/tests/webdriver/bidi/websocket_upgrade.py b/testing/web-platform/mozilla/tests/webdriver/bidi/websocket_upgrade.py
new file mode 100644
index 0000000000..e5ebfa1eb0
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/bidi/websocket_upgrade.py
@@ -0,0 +1,156 @@
+import pytest
+from support.network import get_host, websocket_request
+
+
+@pytest.mark.parametrize(
+ "hostname, port_type, status",
+ [
+ # Valid hosts
+ ("localhost", "server_port", 101),
+ ("localhost", "default_port", 101),
+ ("127.0.0.1", "server_port", 101),
+ ("127.0.0.1", "default_port", 101),
+ ("[::1]", "server_port", 101),
+ ("[::1]", "default_port", 101),
+ ("192.168.8.1", "server_port", 101),
+ ("192.168.8.1", "default_port", 101),
+ ("[fdf8:f535:82e4::53]", "server_port", 101),
+ ("[fdf8:f535:82e4::53]", "default_port", 101),
+ # Invalid hosts
+ ("mozilla.org", "server_port", 400),
+ ("mozilla.org", "wrong_port", 400),
+ ("mozilla.org", "default_port", 400),
+ ("localhost", "wrong_port", 400),
+ ("127.0.0.1", "wrong_port", 400),
+ ("[::1]", "wrong_port", 400),
+ ("192.168.8.1", "wrong_port", 400),
+ ("[fdf8:f535:82e4::53]", "wrong_port", 400),
+ ],
+ ids=[
+ # Valid hosts
+ "localhost with same port as RemoteAgent",
+ "localhost with default port",
+ "127.0.0.1 (loopback) with same port as RemoteAgent",
+ "127.0.0.1 (loopback) with default port",
+ "[::1] (ipv6 loopback) with same port as RemoteAgent",
+ "[::1] (ipv6 loopback) with default port",
+ "ipv4 address with same port as RemoteAgent",
+ "ipv4 address with default port",
+ "ipv6 address with same port as RemoteAgent",
+ "ipv6 address with default port",
+ # Invalid hosts
+ "random hostname with the same port as RemoteAgent",
+ "random hostname with a different port than RemoteAgent",
+ "random hostname with default port",
+ "localhost with a different port than RemoteAgent",
+ "127.0.0.1 (loopback) with a different port than RemoteAgent",
+ "[::1] (ipv6 loopback) with a different port than RemoteAgent",
+ "ipv4 address with a different port than RemoteAgent",
+ "ipv6 address with a different port than RemoteAgent",
+ ],
+)
+def test_host_header(browser, hostname, port_type, status):
+ # Request a default browser
+ current_browser = browser(use_bidi=True)
+ server_host = current_browser.remote_agent_host
+ server_port = current_browser.remote_agent_port
+ test_host = get_host(port_type, hostname, server_port)
+
+ response = websocket_request(server_host, server_port, host=test_host)
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "hostname, port_type, status",
+ [
+ # Allowed hosts
+ ("testhost", "server_port", 101),
+ ("testhost", "default_port", 101),
+ ("testhost", "wrong_port", 400),
+ # IP addresses
+ ("192.168.8.1", "server_port", 101),
+ ("192.168.8.1", "default_port", 101),
+ ("[fdf8:f535:82e4::53]", "server_port", 101),
+ ("[fdf8:f535:82e4::53]", "default_port", 101),
+ ("127.0.0.1", "server_port", 101),
+ ("127.0.0.1", "default_port", 101),
+ ("[::1]", "server_port", 101),
+ ("[::1]", "default_port", 101),
+ # Localhost
+ ("localhost", "server_port", 400),
+ ("localhost", "default_port", 400),
+ ],
+ ids=[
+ # Allowed hosts
+ "allowed host with same port as RemoteAgent",
+ "allowed host with default port",
+ "allowed host with wrong port",
+ # IP addresses
+ "ipv4 address with same port as RemoteAgent",
+ "ipv4 address with default port",
+ "ipv6 address with same port as RemoteAgent",
+ "ipv6 address with default port",
+ "127.0.0.1 (loopback) with same port as RemoteAgent",
+ "127.0.0.1 (loopback) with default port",
+ "[::1] (ipv6 loopback) with same port as RemoteAgent",
+ "[::1] (ipv6 loopback) with default port",
+ # Localhost
+ "localhost with same port as RemoteAgent",
+ "localhost with default port",
+ ],
+)
+def test_allowed_hosts(browser, hostname, port_type, status):
+ # Request a browser with custom allowed hosts.
+ current_browser = browser(
+ use_bidi=True,
+ extra_args=["--remote-allow-hosts", "testhost"],
+ )
+ server_host = current_browser.remote_agent_host
+ server_port = current_browser.remote_agent_port
+ test_host = get_host(port_type, hostname, server_port)
+
+ response = websocket_request(server_host, server_port, host=test_host)
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "origin, status",
+ [
+ (None, 101),
+ ("", 400),
+ ("sometext", 400),
+ ("http://localhost:1234", 400),
+ ],
+)
+def test_origin_header(browser, origin, status):
+ # Request a default browser.
+ current_browser = browser(use_bidi=True)
+ server_host = current_browser.remote_agent_host
+ server_port = current_browser.remote_agent_port
+ response = websocket_request(server_host, server_port, origin=origin)
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "origin, status",
+ [
+ (None, 101),
+ ("", 400),
+ ("sometext", 400),
+ ("http://localhost:1234", 101),
+ ("https://localhost:1234", 400),
+ ],
+)
+def test_allowed_origins(browser, origin, status):
+ # Request a browser with custom allowed origins.
+ current_browser = browser(
+ use_bidi=True,
+ extra_args=["--remote-allow-origins", "http://localhost:1234"],
+ )
+ server_port = current_browser.remote_agent_port
+
+ # Both `localhost` and `127.0.0.1` have to accept connections.
+ for target_host in ["127.0.0.1", "localhost"]:
+ print(f"Connecting to the WebSocket via host {target_host}")
+ response = websocket_request(target_host, server_port, origin=origin)
+ assert response.status == status
diff --git a/testing/web-platform/mozilla/tests/webdriver/cdp/__init__.py b/testing/web-platform/mozilla/tests/webdriver/cdp/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/cdp/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/cdp/debugger_address.py b/testing/web-platform/mozilla/tests/webdriver/cdp/debugger_address.py
new file mode 100644
index 0000000000..ef9a301d24
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/cdp/debugger_address.py
@@ -0,0 +1,45 @@
+# META: timeout=long
+
+import json
+
+import pytest
+from support.context import using_context
+from tests.support.http_request import HTTPRequest
+
+
+def test_debugger_address_not_set(session):
+ debugger_address = session.capabilities.get("moz:debuggerAddress")
+ assert debugger_address is None
+
+
+@pytest.mark.capabilities({"moz:debuggerAddress": False})
+def test_debugger_address_false(session):
+ debugger_address = session.capabilities.get("moz:debuggerAddress")
+ assert debugger_address is None
+
+
+@pytest.mark.capabilities({"moz:debuggerAddress": True})
+@pytest.mark.parametrize("fission_enabled", [True, False], ids=["enabled", "disabled"])
+def test_debugger_address_true_with_fission(session, fission_enabled):
+ debugger_address = session.capabilities.get("moz:debuggerAddress")
+ assert debugger_address is not None
+
+ host, port = debugger_address.split(":")
+ assert host == "127.0.0.1"
+ assert port.isnumeric()
+
+ # Fetch the browser version via the debugger address, `localhost` has
+ # to work as well.
+ for target_host in [host, "localhost"]:
+ print(f"Connecting to WebSocket via host {target_host}")
+ http = HTTPRequest(target_host, int(port))
+ with http.get("/json/version") as response:
+ data = json.loads(response.read())
+ assert session.capabilities["browserVersion"] in data["Browser"]
+
+ # Ensure Fission is not disabled (bug 1813981)
+ with using_context(session, "chrome"):
+ assert (
+ session.execute_script("""return Services.appinfo.fissionAutostart""")
+ is fission_enabled
+ )
diff --git a/testing/web-platform/mozilla/tests/webdriver/cdp/port_file.py b/testing/web-platform/mozilla/tests/webdriver/cdp/port_file.py
new file mode 100644
index 0000000000..aa294deb24
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/cdp/port_file.py
@@ -0,0 +1,30 @@
+import os
+
+from support.network import websocket_request
+
+
+def test_devtools_active_port_file(browser):
+ current_browser = browser(use_cdp=True)
+
+ assert current_browser.remote_agent_port != 0
+ assert current_browser.debugger_address.startswith("/devtools/browser/")
+
+ port_file = os.path.join(current_browser.profile.profile, "DevToolsActivePort")
+ assert os.path.exists(port_file)
+
+ current_browser.quit(clean_profile=False)
+ assert not os.path.exists(port_file)
+
+
+def test_connect(browser):
+ current_browser = browser(use_cdp=True)
+
+ # Both `localhost` and `127.0.0.1` have to accept connections.
+ for target_host in ["127.0.0.1", "localhost"]:
+ print(f"Connecting to the WebSocket via host {target_host}")
+ response = websocket_request(
+ target_host,
+ current_browser.remote_agent_port,
+ path=current_browser.debugger_address,
+ )
+ assert response.status == 101
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/scroll_into_view.py b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/scroll_into_view.py
new file mode 100644
index 0000000000..080195d345
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/scroll_into_view.py
@@ -0,0 +1,50 @@
+from tests.support.asserts import assert_success
+from tests.support.helpers import is_element_in_viewport
+
+
+def element_send_keys(session, element, text):
+ return session.transport.send(
+ "POST",
+ "/session/{session_id}/element/{element_id}/value".format(
+ session_id=session.session_id, element_id=element.id
+ ),
+ {"text": text},
+ )
+
+
+def test_option_select_container_outside_of_scrollable_viewport(session, inline):
+ session.url = inline(
+ """
+ <select style="margin-top: 102vh;">
+ <option value="foo">foo</option>
+ <option value="bar" id="bar">bar</option>
+ </select>
+ """
+ )
+ element = session.find.css("option#bar", all=False)
+ select = session.find.css("select", all=False)
+
+ response = element_send_keys(session, element, "bar")
+ assert_success(response)
+
+ assert is_element_in_viewport(session, select)
+ assert is_element_in_viewport(session, element)
+
+
+def test_option_stays_outside_of_scrollable_viewport(session, inline):
+ session.url = inline(
+ """
+ <select multiple style="height: 105vh; margin-top: 100vh;">
+ <option value="foo" id="foo" style="height: 100vh;">foo</option>
+ <option value="bar" id="bar" style="background-color: yellow;">bar</option>
+ </select>
+ """
+ )
+ select = session.find.css("select", all=False)
+ option_bar = session.find.css("option#bar", all=False)
+
+ response = element_send_keys(session, option_bar, "bar")
+ assert_success(response)
+
+ assert is_element_in_viewport(session, select)
+ assert is_element_in_viewport(session, option_bar)
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/send_keys.py b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/send_keys.py
new file mode 100644
index 0000000000..1399959ceb
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/element_send_keys/send_keys.py
@@ -0,0 +1,24 @@
+from tests.support.asserts import assert_success
+from tests.support.keys import Keys
+
+
+def element_send_keys(session, element, text):
+ return session.transport.send(
+ "POST",
+ "/session/{session_id}/element/{element_id}/value".format(
+ session_id=session.session_id, element_id=element.id
+ ),
+ {"text": text},
+ )
+
+
+def test_modifier_key_toggles(session, inline, modifier_key):
+ session.url = inline("<input type=text value=foo>")
+ element = session.find.css("input", all=False)
+
+ response = element_send_keys(
+ session, element, f"{modifier_key}a{modifier_key}{Keys.DELETE}cheese"
+ )
+ assert_success(response)
+
+ assert element.property("value") == "cheese"
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/execute_async.py b/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/execute_async.py
new file mode 100644
index 0000000000..990f2c1b31
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/execute_async_script/execute_async.py
@@ -0,0 +1,59 @@
+import pytest
+from tests.support.asserts import assert_success
+from tests.support.sync import Poll
+
+
+def execute_async_script(session, script, args=None):
+ if args is None:
+ args = []
+ body = {"script": script, "args": args}
+
+ return session.transport.send(
+ "POST", "/session/{session_id}/execute/async".format(**vars(session)), body
+ )
+
+
+@pytest.mark.parametrize("dialog_type", ["alert", "confirm", "prompt"])
+def test_no_abort_by_user_prompt_in_other_tab(session, inline, dialog_type):
+ original_handle = session.window_handle
+ original_handles = session.handles
+
+ session.url = inline(
+ """
+ <a onclick="window.open();">open window</a>
+ <script>
+ window.addEventListener("message", function (event) {{
+ {}("foo");
+ }});
+ </script>
+ """.format(
+ dialog_type
+ )
+ )
+
+ session.find.css("a", all=False).click()
+ wait = Poll(session, timeout=5, message="No new window has been opened")
+ new_handles = wait.until(lambda s: set(s.handles) - set(original_handles))
+ assert len(new_handles) == 1
+
+ session.window_handle = new_handles.pop()
+
+ response = execute_async_script(
+ session,
+ """
+ const resolve = arguments[0];
+
+ // Trigger opening a user prompt in the other window.
+ window.opener.postMessage("foo", "*");
+
+ // Delay resolving the Promise to ensure a user prompt has been opened.
+ setTimeout(() => resolve(42), 500);
+ """,
+ )
+
+ assert_success(response, 42)
+
+ session.window.close()
+
+ session.window_handle = original_handle
+ session.alert.accept()
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/chrome.py b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/chrome.py
new file mode 100644
index 0000000000..af24be4b9e
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handle/chrome.py
@@ -0,0 +1,25 @@
+from support.context import using_context
+from tests.support.asserts import assert_success
+
+
+def get_window_handle(session):
+ return session.transport.send(
+ "GET", "session/{session_id}/window".format(**vars(session))
+ )
+
+
+def test_basic(session):
+ with using_context(session, "chrome"):
+ response = get_window_handle(session)
+ assert_success(response, session.window_handle)
+
+
+def test_different_handle_than_content_scope(session):
+ response = get_window_handle(session)
+ content_handle = assert_success(response)
+
+ with using_context(session, "chrome"):
+ response = get_window_handle(session)
+ chrome_handle = assert_success(response)
+
+ assert chrome_handle != content_handle
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/chrome.py b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/chrome.py
new file mode 100644
index 0000000000..091ac01e6c
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/get_window_handles/chrome.py
@@ -0,0 +1,43 @@
+from support.context import using_context
+from tests.support.asserts import assert_success
+
+
+def get_window_handles(session):
+ return session.transport.send(
+ "GET", "session/{session_id}/window/handles".format(**vars(session))
+ )
+
+
+def test_basic(session):
+ with using_context(session, "chrome"):
+ response = get_window_handles(session)
+ assert_success(response, session.handles)
+
+
+def test_different_handles_than_content_scope(session):
+ response = get_window_handles(session)
+ content_handles = assert_success(response)
+
+ with using_context(session, "chrome"):
+ response = get_window_handles(session)
+ chrome_handles = assert_success(response)
+
+ assert chrome_handles != content_handles
+ assert len(chrome_handles) == 1
+ assert len(content_handles) == 1
+
+
+def test_multiple_windows_and_tabs(session):
+ session.new_window(type_hint="window")
+ session.new_window(type_hint="tab")
+
+ response = get_window_handles(session)
+ content_handles = assert_success(response)
+
+ with using_context(session, "chrome"):
+ response = get_window_handles(session)
+ chrome_handles = assert_success(response)
+
+ assert chrome_handles != content_handles
+ assert len(chrome_handles) == 2
+ assert len(content_handles) == 3
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/bidi_disabled.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/bidi_disabled.py
new file mode 100644
index 0000000000..eeb5a18740
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/bidi_disabled.py
@@ -0,0 +1,33 @@
+from copy import deepcopy
+
+
+def test_marionette_fallback_webdriver_session(configuration, geckodriver):
+ config = deepcopy(configuration)
+ config["capabilities"]["webSocketUrl"] = True
+
+ prefs = config["capabilities"]["moz:firefoxOptions"].get("prefs", {})
+ prefs.update({"remote.active-protocols": 2})
+ config["capabilities"]["moz:firefoxOptions"]["prefs"] = prefs
+
+ try:
+ driver = geckodriver(config=config)
+ driver.new_session()
+
+ assert driver.session.capabilities.get("webSocketUrl") is None
+
+ # Sanity check that Marionette works as expected and by default returns
+ # at least one window handle
+ assert len(driver.session.handles) >= 1
+
+ finally:
+ driver.stop()
+
+ # WebDriver BiDi has to be re-enabled. Because we cannot easily
+ # get rid of the value let geckodriver overwrite it with the current
+ # default.
+ prefs.update({"remote.active-protocols": 3})
+
+ driver = geckodriver(config=config)
+ driver.new_session()
+
+ assert driver.session.capabilities.get("webSocketUrl") is not None
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/binary.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/binary.py
new file mode 100644
index 0000000000..79d1f842ed
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/binary.py
@@ -0,0 +1,33 @@
+import os
+
+from tests.support.asserts import assert_error, assert_success
+
+
+def test_bad_binary(new_session, configuration):
+ # skipif annotations are forbidden in wpt
+ if os.path.exists("/bin/echo"):
+ capabilities = configuration["capabilities"].copy()
+ capabilities["moz:firefoxOptions"]["binary"] = "/bin/echo"
+
+ response, _ = new_session({"capabilities": {"alwaysMatch": capabilities}})
+ assert_error(response, "invalid argument")
+
+
+def test_shell_script_binary(new_session, configuration):
+ # skipif annotations are forbidden in wpt
+ if os.path.exists("/bin/bash"):
+ capabilities = configuration["capabilities"].copy()
+ binary = configuration["browser"]["binary"]
+
+ path = os.path.abspath("firefox.sh")
+ assert not os.path.exists(path)
+ try:
+ script = f"""#!/bin/bash\n\n"{binary}" $@\n"""
+ with open("firefox.sh", "w") as f:
+ f.write(script)
+ os.chmod(path, 0o744)
+ capabilities["moz:firefoxOptions"]["binary"] = path
+ response, _ = new_session({"capabilities": {"alwaysMatch": capabilities}})
+ assert_success(response)
+ finally:
+ os.unlink(path)
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/conftest.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/conftest.py
new file mode 100644
index 0000000000..1cab6784c2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/conftest.py
@@ -0,0 +1,58 @@
+import pytest
+from webdriver.transport import HTTPWireProtocol
+
+
+@pytest.fixture(name="configuration")
+def fixture_configuration(configuration):
+ """Remove "acceptInsecureCerts" from capabilities if it exists.
+
+ Some browser configurations add acceptInsecureCerts capability by default.
+ Remove it during new_session tests to avoid interference.
+ """
+ if "acceptInsecureCerts" in configuration["capabilities"]:
+ configuration = dict(configuration)
+ del configuration["capabilities"]["acceptInsecureCerts"]
+ return configuration
+
+
+@pytest.fixture(name="new_session")
+def fixture_new_session(request, configuration, current_session):
+ """Start a new session for tests which themselves test creating new sessions.
+
+ :param body: The content of the body for the new session POST request.
+
+ :param delete_existing_session: Allows the fixture to delete an already
+ created custom session before the new session is getting created. This
+ is useful for tests which call this fixture multiple times within the
+ same test.
+ """
+ custom_session = {}
+
+ transport = HTTPWireProtocol(
+ configuration["host"],
+ configuration["port"],
+ url_prefix="/",
+ )
+
+ def _delete_session(session_id):
+ transport.send("DELETE", "session/{}".format(session_id))
+
+ def new_session(body, delete_existing_session=False, headers=None):
+ # If there is an active session from the global session fixture,
+ # delete that one first
+ if current_session is not None:
+ current_session.end()
+
+ if delete_existing_session:
+ _delete_session(custom_session["session"]["sessionId"])
+
+ response = transport.send("POST", "session", body, headers=headers)
+ if response.status == 200:
+ custom_session["session"] = response.body["value"]
+ return response, custom_session.get("session", None)
+
+ yield new_session
+
+ if custom_session.get("session") is not None:
+ _delete_session(custom_session["session"]["sessionId"])
+ custom_session = None
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/create.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/create.py
new file mode 100644
index 0000000000..9649b938ad
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/create.py
@@ -0,0 +1,11 @@
+# META: timeout=long
+from tests.support.asserts import assert_success
+
+
+def test_valid_content_type(new_session, configuration):
+ headers = {"content-type": "application/json"}
+ response, _ = new_session(
+ {"capabilities": {"alwaysMatch": dict(configuration["capabilities"])}},
+ headers=headers,
+ )
+ assert_success(response)
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/invalid.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/invalid.py
new file mode 100644
index 0000000000..dc7a0caee9
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/invalid.py
@@ -0,0 +1,53 @@
+from copy import deepcopy
+
+import pytest
+from tests.support.asserts import assert_error
+
+
+@pytest.mark.parametrize(
+ "headers",
+ [
+ {"origin": "http://localhost"},
+ {"origin": "http://localhost:8000"},
+ {"origin": "http://127.0.0.1"},
+ {"origin": "http://127.0.0.1:8000"},
+ {"origin": "null"},
+ {"ORIGIN": "https://example.org"},
+ {"host": "example.org:4444"},
+ {"Host": "example.org"},
+ {"host": "localhost:80"},
+ {"host": "localhost"},
+ {"content-type": "application/x-www-form-urlencoded"},
+ {"content-type": "multipart/form-data"},
+ {"content-type": "text/plain"},
+ {"Content-TYPE": "APPLICATION/x-www-form-urlencoded"},
+ {"content-type": "MULTIPART/FORM-DATA"},
+ {"CONTENT-TYPE": "TEXT/PLAIN"},
+ {"content-type": "text/plain ; charset=utf-8"},
+ {"content-type": "text/plain;foo"},
+ {"content-type": "text/PLAIN ; foo;charset=utf8"},
+ ],
+)
+def test_invalid(new_session, configuration, headers):
+ response, _ = new_session(
+ {"capabilities": {"alwaysMatch": dict(configuration["capabilities"])}},
+ headers=headers,
+ )
+ assert_error(response, "unknown error")
+
+
+@pytest.mark.parametrize(
+ "argument",
+ [
+ "--marionette",
+ "--remote-debugging-port",
+ "--remote-allow-hosts",
+ "--remote-allow-origins",
+ ],
+)
+def test_forbidden_arguments(configuration, new_session, argument):
+ capabilities = deepcopy(configuration["capabilities"])
+ capabilities["moz:firefoxOptions"]["args"] = [argument]
+
+ response, _ = new_session({"capabilities": {"alwaysMatch": capabilities}})
+ assert_error(response, "invalid argument")
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/new_session/profile_root.py b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/profile_root.py
new file mode 100644
index 0000000000..8bbdbadc69
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/new_session/profile_root.py
@@ -0,0 +1,38 @@
+import copy
+import os
+
+import pytest
+
+
+def test_profile_root(tmp_path, configuration, geckodriver):
+ profile_path = os.path.join(tmp_path, "geckodriver-test")
+ os.makedirs(profile_path)
+
+ config = copy.deepcopy(configuration)
+ # Ensure we don't set a profile in command line arguments
+ del config["capabilities"]["moz:firefoxOptions"]["args"]
+
+ extra_args = ["--profile-root", profile_path]
+
+ assert os.listdir(profile_path) == []
+
+ driver = geckodriver(config=config, extra_args=extra_args)
+ driver.new_session()
+ assert len(os.listdir(profile_path)) == 1
+ driver.delete_session()
+ assert os.listdir(profile_path) == []
+
+
+def test_profile_root_missing(tmp_path, configuration, geckodriver):
+ profile_path = os.path.join(tmp_path, "missing-path")
+ assert not os.path.exists(profile_path)
+
+ config = copy.deepcopy(configuration)
+ # Ensure we don't set a profile in command line arguments
+ del config["capabilities"]["moz:firefoxOptions"]["args"]
+
+ extra_args = ["--profile-root", profile_path]
+
+ with pytest.raises(ChildProcessError) as exc_info:
+ geckodriver(config=config, extra_args=extra_args)
+ assert str(exc_info.value) == "geckodriver terminated with code 64"
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/protocol/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_hosts.py b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_hosts.py
new file mode 100644
index 0000000000..17ae2c2c68
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_hosts.py
@@ -0,0 +1,53 @@
+from copy import deepcopy
+
+import pytest
+from support.network import get_host, http_request, websocket_request
+
+
+@pytest.mark.parametrize(
+ "allow_hosts, hostname, port_type, status",
+ [
+ # Valid hosts
+ (["localhost.localdomain", "localhost"], "localhost", "server_port", 200),
+ (["localhost.localdomain", "localhost"], "127.0.0.1", "server_port", 200),
+ # Invalid hosts
+ (["localhost.localdomain"], "localhost", "server_port", 500),
+ (["localhost"], "localhost", "wrong_port", 500),
+ (["www.localhost"], "localhost", "server_port", 500),
+ ],
+)
+def test_allow_hosts(geckodriver, allow_hosts, hostname, port_type, status):
+ extra_args = ["--allow-hosts"] + allow_hosts
+
+ driver = geckodriver(hostname=hostname, extra_args=extra_args)
+ host = get_host(port_type, hostname, driver.port)
+ response = http_request(driver.hostname, driver.port, host=host)
+
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "allow_hosts, hostname, status",
+ [
+ (["mozilla.org", "testhost"], "testhost", 101),
+ (["mozilla.org"], "testhost", 400),
+ ],
+ ids=["allowed", "not allowed"],
+)
+def test_allow_hosts_passed_to_remote_agent(
+ configuration, geckodriver, allow_hosts, hostname, status
+):
+ config = deepcopy(configuration)
+ config["capabilities"]["webSocketUrl"] = True
+
+ extra_args = ["--allow-hosts"] + allow_hosts
+
+ driver = geckodriver(config=config, extra_args=extra_args)
+
+ driver.new_session()
+
+ host = get_host("default_port", hostname, driver.remote_agent_port)
+ response = websocket_request("127.0.0.1", driver.remote_agent_port, host=host)
+ assert response.status == status
+
+ driver.delete_session()
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_origins.py b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_origins.py
new file mode 100644
index 0000000000..72b6fba482
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/allow_origins.py
@@ -0,0 +1,56 @@
+from copy import deepcopy
+
+import pytest
+from support.network import http_request, websocket_request
+
+
+@pytest.mark.parametrize(
+ "allow_origins, origin, status",
+ [
+ # Valid origins
+ (["http://web-platform.test"], "http://web-platform.test", 200),
+ (["http://web-platform.test"], "http://web-platform.test:80", 200),
+ (["https://web-platform.test"], "https://web-platform.test:443", 200),
+ # Invalid origins
+ (["https://web-platform.test"], "http://web-platform.test", 500),
+ (["http://web-platform.test:8000"], "http://web-platform.test", 500),
+ (["http://web-platform.test"], "http://www.web-platform.test", 500),
+ ],
+)
+def test_allow_hosts(configuration, geckodriver, allow_origins, origin, status):
+ extra_args = ["--allow-origins"] + allow_origins
+
+ driver = geckodriver(extra_args=extra_args)
+ response = http_request(driver.hostname, driver.port, origin=origin)
+
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "allow_origins, origin, status",
+ [
+ (
+ ["https://web-platform.test", "http://web-platform.test"],
+ "http://web-platform.test",
+ 101,
+ ),
+ (["https://web-platform.test"], "http://web-platform.test", 400),
+ ],
+ ids=["allowed", "not allowed"],
+)
+def test_allow_origins_passed_to_remote_agent(
+ configuration, geckodriver, allow_origins, origin, status
+):
+ config = deepcopy(configuration)
+ config["capabilities"]["webSocketUrl"] = True
+
+ extra_args = ["--allow-origins"] + allow_origins
+
+ driver = geckodriver(config=config, extra_args=extra_args)
+
+ driver.new_session()
+
+ response = websocket_request("127.0.0.1", driver.remote_agent_port, origin=origin)
+ assert response.status == status
+
+ driver.delete_session()
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/protocol/marionette_port.py b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/marionette_port.py
new file mode 100644
index 0000000000..09951abc43
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/marionette_port.py
@@ -0,0 +1,41 @@
+import os
+from copy import deepcopy
+
+import pytest
+
+
+@pytest.mark.parametrize("port", ["0", "2828"], ids=["system allocated", "fixed"])
+def test_marionette_port(geckodriver, port):
+ extra_args = ["--marionette-port", port]
+
+ driver = geckodriver(extra_args=extra_args)
+ driver.new_session()
+ driver.delete_session()
+
+
+def test_marionette_port_outdated_active_port_file(
+ configuration, geckodriver, custom_profile
+):
+ config = deepcopy(configuration)
+ extra_args = ["--marionette-port", "0"]
+
+ # Prepare a Marionette active port file that contains a port which will
+ # never be used when requesting a system allocated port.
+ active_port_file = os.path.join(custom_profile.profile, "MarionetteActivePort")
+ with open(active_port_file, "wb") as f:
+ f.write(b"53")
+
+ config["capabilities"]["moz:firefoxOptions"]["args"] = [
+ "--profile",
+ custom_profile.profile,
+ ]
+
+ driver = geckodriver(config=config, extra_args=extra_args)
+
+ driver.new_session()
+ with open(active_port_file, "rb") as f:
+ assert f.readline() != b"53"
+
+ driver.delete_session()
+ with pytest.raises(FileNotFoundError):
+ open(active_port_file, "rb")
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/protocol/request.py b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/request.py
new file mode 100644
index 0000000000..ad99d6964d
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/protocol/request.py
@@ -0,0 +1,72 @@
+import pytest
+from support.network import get_host, http_request
+
+
+@pytest.mark.parametrize(
+ "hostname, port_type, status",
+ [
+ # Valid hosts
+ ("localhost", "server_port", 200),
+ ("127.0.0.1", "server_port", 200),
+ ("[::1]", "server_port", 200),
+ ("192.168.8.1", "server_port", 200),
+ ("[fdf8:f535:82e4::53]", "server_port", 200),
+ # Invalid hosts
+ ("localhost", "default_port", 500),
+ ("127.0.0.1", "default_port", 500),
+ ("[::1]", "default_port", 500),
+ ("192.168.8.1", "default_port", 500),
+ ("[fdf8:f535:82e4::53]", "default_port", 500),
+ ("example.org", "server_port", 500),
+ ("example.org", "wrong_port", 500),
+ ("example.org", "default_port", 500),
+ ("localhost", "wrong_port", 500),
+ ("127.0.0.1", "wrong_port", 500),
+ ("[::1]", "wrong_port", 500),
+ ("192.168.8.1", "wrong_port", 500),
+ ("[fdf8:f535:82e4::53]", "wrong_port", 500),
+ ],
+ ids=[
+ # Valid hosts
+ "localhost with same port as server",
+ "127.0.0.1 (loopback) with same port as server",
+ "[::1] (ipv6 loopback) with same port as server",
+ "ipv4 address with same port as server",
+ "ipv6 address with same port as server",
+ # Invalid hosts
+ "localhost with default port",
+ "127.0.0.1 (loopback) with default port",
+ "[::1] (ipv6 loopback) with default port",
+ "ipv4 address with default port",
+ "ipv6 address with default port",
+ "random hostname with the same port as server",
+ "random hostname with a different port than server",
+ "random hostname with default port",
+ "localhost with a different port than server",
+ "127.0.0.1 (loopback) with a different port than server",
+ "[::1] (ipv6 loopback) with a different port than server",
+ "ipv4 address with a different port than server",
+ "ipv6 address with a different port than server",
+ ],
+)
+def test_host_header(configuration, hostname, port_type, status):
+ host = get_host(port_type, hostname, configuration["port"])
+ response = http_request(configuration["host"], configuration["port"], host=host)
+
+ assert response.status == status
+
+
+@pytest.mark.parametrize(
+ "origin, add_port, status",
+ [
+ (None, False, 200),
+ ("", False, 500),
+ ("sometext", False, 500),
+ ("http://localhost", True, 500),
+ ],
+)
+def test_origin_header(configuration, origin, add_port, status):
+ if add_port:
+ origin = f"{origin}:{configuration['port']}"
+ response = http_request(configuration["host"], configuration["port"], origin=origin)
+ assert response.status == status
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/send_alert_text.py b/testing/web-platform/mozilla/tests/webdriver/classic/send_alert_text.py
new file mode 100644
index 0000000000..60d6a02af0
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/send_alert_text.py
@@ -0,0 +1,22 @@
+from tests.support.asserts import assert_error
+from tests.support.http_handlers.authentication import basic_authentication
+
+
+def send_alert_text(session, text=None):
+ return session.transport.send(
+ "POST",
+ "session/{session_id}/alert/text".format(**vars(session)),
+ {"text": text},
+ )
+
+
+def test_basic_auth_unsupported_operation(url, session):
+ """
+ Basic auth dialogues are not included in HTML's definition of
+ 'user prompts': those are limited to the 'simple dialogues'
+ such as window.alert(), window.prompt() et al. and the print
+ dialogue.
+ """
+ session.url = basic_authentication(url)
+ response = send_alert_text(session, "Federer")
+ assert_error(response, "unsupported operation")
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/__init__.py b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/__init__.py
new file mode 100644
index 0000000000..11a8a58a0f
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/__init__.py
@@ -0,0 +1,12 @@
+def document_dimensions(session):
+ return tuple(
+ session.execute_script(
+ """
+ const {devicePixelRatio} = window;
+ const width = document.documentElement.scrollWidth;
+ const height = document.documentElement.scrollHeight;
+
+ return [Math.floor(width * devicePixelRatio), Math.floor(height * devicePixelRatio)];
+ """
+ )
+ )
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/iframe.py b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/iframe.py
new file mode 100644
index 0000000000..fc231f2e11
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/iframe.py
@@ -0,0 +1,47 @@
+import pytest
+from tests.support.asserts import assert_success
+from tests.support.image import png_dimensions
+
+from . import document_dimensions
+
+DEFAULT_CSS_STYLE = """
+ <style>
+ div, iframe {
+ display: block;
+ border: 1px solid blue;
+ width: 10em;
+ height: 10em;
+ }
+ </style>
+"""
+
+DEFAULT_CONTENT = "<div>Lorem ipsum dolor sit amet.</div>"
+
+
+def take_full_screenshot(session):
+ return session.transport.send(
+ "GET",
+ "/session/{session_id}/moz/screenshot/full".format(
+ session_id=session.session_id
+ ),
+ )
+
+
+@pytest.mark.parametrize("domain", ["", "alt"], ids=["same_origin", "cross_origin"])
+def test_source_origin(session, url, domain, inline, iframe):
+ session.url = inline("""{0}{1}""".format(DEFAULT_CSS_STYLE, DEFAULT_CONTENT))
+
+ response = take_full_screenshot(session)
+ reference_screenshot = assert_success(response)
+ assert png_dimensions(reference_screenshot) == document_dimensions(session)
+
+ iframe_content = "<style>body {{ margin: 0; }}</style>{}".format(DEFAULT_CONTENT)
+ session.url = inline(
+ """{0}{1}""".format(DEFAULT_CSS_STYLE, iframe(iframe_content, domain=domain))
+ )
+
+ response = take_full_screenshot(session)
+ screenshot = assert_success(response)
+ assert png_dimensions(screenshot) == document_dimensions(session)
+
+ assert screenshot == reference_screenshot
diff --git a/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/screenshot.py b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/screenshot.py
new file mode 100644
index 0000000000..02373afd57
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/classic/take_full_screenshot/screenshot.py
@@ -0,0 +1,51 @@
+from tests.support.asserts import assert_error, assert_png, assert_success
+from tests.support.image import png_dimensions
+
+from . import document_dimensions
+
+
+def take_full_screenshot(session):
+ return session.transport.send(
+ "GET",
+ "/session/{session_id}/moz/screenshot/full".format(
+ session_id=session.session_id
+ ),
+ )
+
+
+def test_no_browsing_context(session, closed_window):
+ response = take_full_screenshot(session)
+ assert_error(response, "no such window")
+
+
+def test_html_document(session, inline):
+ session.url = inline("<input>")
+
+ response = take_full_screenshot(session)
+ value = assert_success(response)
+ assert_png(value)
+ assert png_dimensions(value) == document_dimensions(session)
+
+
+def test_xhtml_document(session, inline):
+ session.url = inline('<input type="text" />', doctype="xhtml")
+
+ response = take_full_screenshot(session)
+ value = assert_success(response)
+ assert_png(value)
+ assert png_dimensions(value) == document_dimensions(session)
+
+
+def test_document_extends_beyond_viewport(session, inline):
+ session.url = inline(
+ """
+ <style>
+ body { min-height: 200vh }
+ </style>
+ """
+ )
+
+ response = take_full_screenshot(session)
+ value = assert_success(response)
+ assert_png(value)
+ assert png_dimensions(value) == document_dimensions(session)
diff --git a/testing/web-platform/mozilla/tests/webdriver/conftest.py b/testing/web-platform/mozilla/tests/webdriver/conftest.py
new file mode 100644
index 0000000000..d754b39e79
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/conftest.py
@@ -0,0 +1,15 @@
+import os
+import sys
+
+base = os.path.dirname(__file__)
+webdriver_path = os.path.abspath(
+ os.path.join(base, "..", "..", "..", "tests", "webdriver")
+)
+sys.path.insert(0, os.path.join(webdriver_path))
+
+pytest_plugins = [
+ "support.fixtures",
+ "tests.support.fixtures",
+ "tests.support.fixtures_bidi",
+ "tests.support.fixtures_http",
+]
diff --git a/testing/web-platform/mozilla/tests/webdriver/support/__init__.py b/testing/web-platform/mozilla/tests/webdriver/support/__init__.py
new file mode 100644
index 0000000000..e69de29bb2
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/support/__init__.py
diff --git a/testing/web-platform/mozilla/tests/webdriver/support/context.py b/testing/web-platform/mozilla/tests/webdriver/support/context.py
new file mode 100644
index 0000000000..910b202075
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/support/context.py
@@ -0,0 +1,20 @@
+import contextlib
+
+
+def set_context(session, context):
+ session.send_session_command("POST", "moz/context", {"context": context})
+
+
+@contextlib.contextmanager
+def using_context(session, context):
+ orig_context = session.send_session_command("GET", "moz/context")
+ needs_change = context != orig_context
+
+ if needs_change:
+ set_context(session, context)
+
+ try:
+ yield
+ finally:
+ if needs_change:
+ set_context(session, orig_context)
diff --git a/testing/web-platform/mozilla/tests/webdriver/support/fixtures.py b/testing/web-platform/mozilla/tests/webdriver/support/fixtures.py
new file mode 100644
index 0000000000..450f662969
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/support/fixtures.py
@@ -0,0 +1,256 @@
+import json
+import os
+import socket
+import subprocess
+import time
+from contextlib import suppress
+from urllib.parse import urlparse
+
+import pytest
+import webdriver
+from mozprofile import Profile
+from mozrunner import FirefoxRunner
+
+from support.network import get_free_port
+
+
+@pytest.fixture(scope="module")
+def browser(full_configuration):
+ """Start a Firefox instance without using geckodriver.
+
+ geckodriver will automatically use the --remote-allow-hosts and
+ --remote.allow.origins command line arguments.
+
+ Starting Firefox without geckodriver allows to set those command line arguments
+ as needed. The fixture method returns the browser instance that should be used
+ to connect to a RemoteAgent supported protocol (CDP, WebDriver BiDi).
+ """
+ current_browser = None
+
+ def _browser(use_bidi=False, use_cdp=False, extra_args=None, extra_prefs=None):
+ nonlocal current_browser
+
+ # If the requested preferences and arguments match the ones for the
+ # already started firefox, we can reuse the current firefox instance,
+ # return the instance immediately.
+ if current_browser:
+ if (
+ current_browser.use_bidi == use_bidi
+ and current_browser.use_cdp == use_cdp
+ and current_browser.extra_args == extra_args
+ and current_browser.extra_prefs == extra_prefs
+ and current_browser.is_running
+ ):
+ return current_browser
+
+ # Otherwise, if firefox is already started, terminate it because we need
+ # to create a new instance for the provided preferences.
+ current_browser.quit()
+
+ binary = full_configuration["browser"]["binary"]
+ firefox_options = full_configuration["capabilities"]["moz:firefoxOptions"]
+ current_browser = Browser(
+ binary,
+ firefox_options,
+ use_bidi=use_bidi,
+ use_cdp=use_cdp,
+ extra_args=extra_args,
+ extra_prefs=extra_prefs,
+ )
+ current_browser.start()
+ return current_browser
+
+ yield _browser
+
+ # Stop firefox at the end of the test module.
+ if current_browser is not None:
+ current_browser.quit()
+ current_browser = None
+
+
+@pytest.fixture
+def custom_profile(configuration):
+ # Clone the known profile for automation preferences
+ firefox_options = configuration["capabilities"]["moz:firefoxOptions"]
+ _, profile_folder = firefox_options["args"]
+ profile = Profile.clone(profile_folder)
+
+ yield profile
+
+ profile.cleanup()
+
+
+@pytest.fixture
+def geckodriver(configuration):
+ """Start a geckodriver instance directly."""
+ driver = None
+
+ def _geckodriver(config=None, hostname=None, extra_args=None):
+ nonlocal driver
+
+ if config is None:
+ config = configuration
+
+ driver = Geckodriver(config, hostname, extra_args)
+ driver.start()
+
+ return driver
+
+ yield _geckodriver
+
+ if driver is not None:
+ driver.stop()
+
+
+class Browser:
+ def __init__(
+ self,
+ binary,
+ firefox_options,
+ use_bidi=False,
+ use_cdp=False,
+ extra_args=None,
+ extra_prefs=None,
+ ):
+ self.use_bidi = use_bidi
+ self.bidi_port_file = None
+ self.use_cdp = use_cdp
+ self.cdp_port_file = None
+ self.extra_args = extra_args
+ self.extra_prefs = extra_prefs
+
+ self.debugger_address = None
+ self.remote_agent_host = None
+ self.remote_agent_port = None
+
+ # Prepare temporary profile
+ _profile_arg, profile_folder = firefox_options["args"]
+ self.profile = Profile.clone(profile_folder)
+ if self.extra_prefs is not None:
+ self.profile.set_preferences(self.extra_prefs)
+
+ if use_cdp:
+ self.cdp_port_file = os.path.join(
+ self.profile.profile, "DevToolsActivePort"
+ )
+ with suppress(FileNotFoundError):
+ os.remove(self.cdp_port_file)
+ if use_bidi:
+ self.webdriver_bidi_file = os.path.join(
+ self.profile.profile, "WebDriverBiDiServer.json"
+ )
+ with suppress(FileNotFoundError):
+ os.remove(self.webdriver_bidi_file)
+
+ cmdargs = ["-no-remote"]
+ if self.use_bidi or self.use_cdp:
+ cmdargs.extend(["--remote-debugging-port", "0"])
+ if self.extra_args is not None:
+ cmdargs.extend(self.extra_args)
+ self.runner = FirefoxRunner(
+ binary=binary, profile=self.profile, cmdargs=cmdargs
+ )
+
+ @property
+ def is_running(self):
+ return self.runner.is_running()
+
+ def start(self):
+ # Start Firefox.
+ self.runner.start()
+
+ if self.use_bidi:
+ # Wait until the WebDriverBiDiServer.json file is ready
+ while not os.path.exists(self.webdriver_bidi_file):
+ time.sleep(0.1)
+
+ # Read the connection details from file
+ data = json.loads(open(self.webdriver_bidi_file).read())
+ self.remote_agent_host = data["ws_host"]
+ self.remote_agent_port = int(data["ws_port"])
+
+ if self.use_cdp:
+ # Wait until the DevToolsActivePort file is ready
+ while not os.path.exists(self.cdp_port_file):
+ time.sleep(0.1)
+
+ # Read the port if needed and the debugger address from the
+ # DevToolsActivePort file
+ lines = open(self.cdp_port_file).readlines()
+ assert len(lines) == 2
+
+ if self.remote_agent_port is None:
+ self.remote_agent_port = int(lines[0].strip())
+ self.debugger_address = lines[1].strip()
+
+ def quit(self, clean_profile=True):
+ if self.is_running:
+ self.runner.stop()
+ self.runner.cleanup()
+
+ if clean_profile:
+ self.profile.cleanup()
+
+
+class Geckodriver:
+ def __init__(self, configuration, hostname=None, extra_args=None):
+ self.config = configuration["webdriver"]
+ self.requested_capabilities = configuration["capabilities"]
+ self.hostname = hostname or configuration["host"]
+ self.extra_args = extra_args or []
+
+ self.command = None
+ self.proc = None
+ self.port = get_free_port()
+
+ capabilities = {"alwaysMatch": self.requested_capabilities}
+ self.session = webdriver.Session(
+ self.hostname, self.port, capabilities=capabilities
+ )
+
+ @property
+ def remote_agent_port(self):
+ webSocketUrl = self.session.capabilities.get("webSocketUrl")
+ assert webSocketUrl is not None
+
+ return urlparse(webSocketUrl).port
+
+ def start(self):
+ self.command = (
+ [self.config["binary"], "--port", str(self.port)]
+ + self.config["args"]
+ + self.extra_args
+ )
+
+ print(f"Running command: {' '.join(self.command)}")
+ self.proc = subprocess.Popen(self.command)
+
+ # Wait for the port to become ready
+ end_time = time.time() + 10
+ while time.time() < end_time:
+ returncode = self.proc.poll()
+ if returncode is not None:
+ raise ChildProcessError(
+ f"geckodriver terminated with code {returncode}"
+ )
+ with socket.socket() as sock:
+ if sock.connect_ex((self.hostname, self.port)) == 0:
+ break
+ else:
+ raise ConnectionRefusedError(
+ f"Failed to connect to geckodriver on {self.hostname}:{self.port}"
+ )
+
+ return self
+
+ def stop(self):
+ self.delete_session()
+
+ if self.proc:
+ self.proc.kill()
+
+ def new_session(self):
+ self.session.start()
+
+ def delete_session(self):
+ self.session.end()
diff --git a/testing/web-platform/mozilla/tests/webdriver/support/network.py b/testing/web-platform/mozilla/tests/webdriver/support/network.py
new file mode 100644
index 0000000000..25492ca5e5
--- /dev/null
+++ b/testing/web-platform/mozilla/tests/webdriver/support/network.py
@@ -0,0 +1,78 @@
+import socket
+from http.client import HTTPConnection
+
+
+def websocket_request(
+ remote_agent_host, remote_agent_port, host=None, origin=None, path="/session"
+):
+ real_host = f"{remote_agent_host}:{remote_agent_port}"
+ url = f"http://{real_host}{path}"
+
+ conn = HTTPConnection(real_host)
+
+ skip_host = host is not None
+ conn.putrequest("GET", url, skip_host)
+
+ if host is not None:
+ conn.putheader("Host", host)
+
+ if origin is not None:
+ conn.putheader("Origin", origin)
+
+ conn.putheader("Connection", "upgrade")
+ conn.putheader("Upgrade", "websocket")
+ conn.putheader("Sec-WebSocket-Key", "dGhlIHNhbXBsZSBub25jZQ==")
+ conn.putheader("Sec-WebSocket-Version", "13")
+
+ conn.endheaders()
+
+ return conn.getresponse()
+
+
+def http_request(server_host, server_port, path="/status", host=None, origin=None):
+ url = f"http://{server_host}:{server_port}{path}"
+
+ conn = HTTPConnection(server_host, server_port)
+
+ custom_host = host is not None
+ conn.putrequest("GET", url, skip_host=custom_host)
+ if custom_host:
+ conn.putheader("Host", host)
+
+ if origin is not None:
+ conn.putheader("Origin", origin)
+
+ conn.endheaders()
+
+ return conn.getresponse()
+
+
+def get_free_port():
+ """Get a random unbound port"""
+ max_attempts = 10
+ err = None
+ for _ in range(max_attempts):
+ s = socket.socket()
+ try:
+ s.bind(("127.0.0.1", 0))
+ except OSError as e:
+ err = e
+ continue
+ else:
+ return s.getsockname()[1]
+ finally:
+ s.close()
+ if err is None:
+ err = Exception("Failed to get a free port")
+ raise err
+
+
+def get_host(port_type, hostname, server_port):
+ if port_type == "default_port":
+ return hostname
+ if port_type == "server_port":
+ return f"{hostname}:{server_port}"
+ if port_type == "wrong_port":
+ wrong_port = int(server_port) + 1
+ return f"{hostname}:{wrong_port}"
+ raise Exception(f"Unrecognised port_type {port_type}")