')
if request.param == 'about:blank':
@@ -241,10 +252,10 @@ def url_example(local_server_http):
@pytest.fixture
-def url_another_example(local_server_http):
+def url_example_another_origin(local_server_http_another_host):
"""Return a generic example URL with status code 200, in a domain other than
the example_url fixture."""
- return local_server_http.url_200('127.0.0.1')
+ return local_server_http_another_host.url_200()
@pytest.fixture
@@ -397,10 +408,10 @@ async def activate_main_tab():
@pytest.fixture
-def html():
- """Return a factory for HTML data URL with the given content."""
+def html(local_server_http):
+ """Return a factory for URL with the given content."""
def html(content=""):
- return f'data:text/html,{content}'
+ return local_server_http.url_200(content=content)
return html
@@ -414,17 +425,11 @@ def iframe(src=""):
return iframe
-@pytest.fixture
-def html_iframe_same_origin(html, iframe, url_same_origin):
- """Return a page URL with an iframe of the same origin."""
- return html(iframe(url_same_origin))
-
-
@pytest_asyncio.fixture
-async def iframe_id(websocket, context_id: str, html_iframe_same_origin, html):
+async def iframe_id(websocket, context_id, html, iframe):
"""Navigate to a page with an iframe of the same origin, and return the
iframe browser context id."""
- await goto_url(websocket, context_id, html_iframe_same_origin)
+ await goto_url(websocket, context_id, html(iframe(html("
FRAME
"))))
result = await get_tree(websocket, context_id)
iframe_id = result["contexts"][0]["children"][0]["context"]
diff --git a/tests/log/test_log_entry_added.py b/tests/log/test_log_entry_added.py
index 2f816f6b4b..87c5638c75 100644
--- a/tests/log/test_log_entry_added.py
+++ b/tests/log/test_log_entry_added.py
@@ -327,11 +327,12 @@ async def test_exceptionThrown_logEntryAddedEventEmitted(
websocket, context_id, html):
await subscribe(websocket, ["log.entryAdded"])
+ url = html("")
await send_JSON_command(
websocket, {
"method": "browsingContext.navigate",
"params": {
- "url": html(""),
+ "url": url,
"wait": "interactive",
"context": context_id
}
@@ -353,10 +354,12 @@ async def test_exceptionThrown_logEntryAddedEventEmitted(
"timestamp": ANY_TIMESTAMP,
"stackTrace": {
"callFrames": [{
- "url": "",
+ "url": url,
"functionName": "",
"lineNumber": 0,
- "columnNumber": 14
+ # Column number is a magical constant. It depends on the
+ # html fixture wrapping content in document tag.
+ "columnNumber": 127
}]
},
# ConsoleLogEntry
diff --git a/tests/network/test_remove_intercept.py b/tests/network/test_remove_intercept.py
index 6f8016123d..d5d868faf6 100644
--- a/tests/network/test_remove_intercept.py
+++ b/tests/network/test_remove_intercept.py
@@ -220,7 +220,7 @@ async def test_remove_intercept_unblocks(websocket, context_id,
@pytest.mark.asyncio
async def test_remove_intercept_does_not_affect_another_intercept(
websocket, context_id, another_context_id, url_example,
- url_another_example):
+ url_example_another_origin):
await subscribe(websocket, ["network.beforeRequestSent"])
result = await execute_command(
@@ -246,7 +246,7 @@ async def test_remove_intercept_does_not_affect_another_intercept(
"phases": ["beforeRequestSent"],
"urlPatterns": [{
"type": "string",
- "pattern": url_another_example,
+ "pattern": url_example_another_origin,
}, ]
},
})
@@ -293,7 +293,7 @@ async def test_remove_intercept_does_not_affect_another_intercept(
websocket, {
"method": "browsingContext.navigate",
"params": {
- "url": url_another_example,
+ "url": url_example_another_origin,
"context": another_context_id,
"wait": "complete",
}
@@ -313,7 +313,7 @@ async def test_remove_intercept_does_not_affect_another_intercept(
"redirectCount": 0,
"request": {
"request": ANY_STR,
- "url": url_another_example,
+ "url": url_example_another_origin,
"method": "GET",
"headers": ANY_LIST,
"cookies": [],
@@ -362,7 +362,7 @@ async def test_remove_intercept_does_not_affect_another_intercept(
"headers": ANY_LIST,
"method": "GET",
"request": network_id_2,
- "url": url_another_example,
+ "url": url_example_another_origin,
}, ),
"timestamp": ANY_TIMESTAMP,
},
diff --git a/tests/script/test_realm.py b/tests/script/test_realm.py
index 6d288ac460..79b76fe9f7 100644
--- a/tests/script/test_realm.py
+++ b/tests/script/test_realm.py
@@ -21,7 +21,8 @@
@pytest.mark.asyncio
-async def test_realm_realmCreated(websocket, context_id, html):
+async def test_realm_realmCreated(websocket, context_id, html,
+ local_server_http):
url = html()
await subscribe(websocket, ["script.realmCreated"])
@@ -43,7 +44,7 @@ async def test_realm_realmCreated(websocket, context_id, html):
"method": "script.realmCreated",
"params": {
"type": "window",
- "origin": "null",
+ "origin": local_server_http.origin(),
"realm": ANY_STR,
"context": context_id,
}
diff --git a/tests/session/test_subscription.py b/tests/session/test_subscription.py
index 165cd00e9c..72cf7ad43c 100644
--- a/tests/session/test_subscription.py
+++ b/tests/session/test_subscription.py
@@ -100,7 +100,7 @@ async def test_subscribeWithContext_subscribesToEventsInGivenContext(
@pytest.mark.asyncio
async def test_subscribeWithContext_subscribesToEventsInNestedContext(
- websocket, context_id, html_iframe_same_origin, url_same_origin):
+ websocket, context_id, html, iframe, url_all_origins):
await subscribe(websocket, ["browsingContext.contextCreated"])
# Navigate to some page.
@@ -108,7 +108,7 @@ async def test_subscribeWithContext_subscribesToEventsInNestedContext(
websocket, {
"method": "browsingContext.navigate",
"params": {
- "url": html_iframe_same_origin,
+ "url": html(iframe(url_all_origins)),
"wait": "complete",
"context": context_id
}
@@ -121,7 +121,9 @@ async def test_subscribeWithContext_subscribesToEventsInNestedContext(
"method": "browsingContext.contextCreated",
"params": {
"context": ANY_STR,
- "url": url_same_origin,
+ # The `url` is always `about:blank`, as the navigation has not
+ # happened yet. https://github.com/w3c/webdriver-bidi/issues/220.
+ "url": "about:blank",
"children": None,
"parent": context_id,
"userContext": "default",
diff --git a/tests/tools/local_http_server.py b/tests/tools/local_http_server.py
index 9123bc7a74..b4186cbc4c 100644
--- a/tests/tools/local_http_server.py
+++ b/tests/tools/local_http_server.py
@@ -15,6 +15,7 @@
import base64
import ssl
+import uuid
from datetime import datetime
from pathlib import Path
from threading import Event
@@ -25,8 +26,12 @@
class LocalHttpServer:
- """A wrapper of `pytest_httpserver.httpserver` to simplify the usage. Sets
- up common use cases and provides url for them."""
+ """
+ A wrapper of `pytest_httpserver.httpserver` to simplify the usage. Sets up
+ common use cases and provides url for them.
+ NOTE: the server does not support concurrent requests to different origins.
+ If needed, use a instance of the server per origin.
+ """
__http_server: HTTPServer
@@ -54,7 +59,12 @@ def stop(self):
self.hang_forever_stop()
self.__http_server.stop()
- def __init__(self, protocol: Literal['http', 'https'] = 'http') -> None:
+ def __html_doc(self, content):
+ return f"{content}"
+
+ def __init__(self,
+ host: str = 'localhost',
+ protocol: Literal['http', 'https'] = 'http') -> None:
super().__init__()
self.__protocol = protocol
@@ -68,20 +78,17 @@ def __init__(self, protocol: Literal['http', 'https'] = 'http') -> None:
elif protocol != 'http':
raise ValueError(f"Unsupported protocol: {protocol}")
- self.__http_server = HTTPServer(ssl_context=ssl_context)
+ self.__http_server = HTTPServer(host=host, ssl_context=ssl_context)
self.__http_server.start()
self.__http_server.clear()
self.__start_time = datetime.now()
- def html_doc(content):
- return f"{content}"
-
self.__http_server \
.expect_request(self.__path_base) \
.respond_with_data(
- html_doc("I prevent CORS"),
+ self.__html_doc("I prevent CORS"),
headers={"Content-Type": "text/html"})
self.__http_server \
@@ -94,7 +101,7 @@ def html_doc(content):
self.__http_server \
.expect_request(self.__path_200) \
.respond_with_data(
- html_doc(self.content_200),
+ self.__html_doc(self.content_200),
headers={"Content-Type": "text/html"})
# Set up permanent redirect.
@@ -135,7 +142,7 @@ def hang_forever(_):
.respond_with_handler(hang_forever)
def cache(request: Request):
- content = html_doc(self.content_200)
+ content = self.__html_doc(self.content_200)
if_modified_since = request.headers.get("If-Modified-Since")
if if_modified_since is not None:
@@ -158,48 +165,44 @@ def hang_forever_stop(self):
if self.hang_forever_stop_flag is not None:
self.hang_forever_stop_flag.set()
- def _url_for(self, suffix: str, host: str = 'localhost') -> str:
- """
- Return an url for a given suffix.
-
- Implementation is the same as the original one, but with a customizable
- host: https://github.com/csernazs/pytest-httpserver/blob/8110d9d543de3b7c151bc1b5c8e85c01b05b226d/pytest_httpserver/httpserver.py#L665
- :param suffix: the suffix which will be added to the base url. It can
- start with ``/`` (slash) or not, the url will be the same.
- :param host: the host to use in the url. Default is ``localhost``.
- :return: the full url which refers to the server
+ def origin(self) -> str:
+ """Returns the url for the base page to navigate and prevent CORS.
"""
- if not suffix.startswith("/"):
- suffix = "/" + suffix
+ return self.url_base()[:-1]
- host = self.__http_server.format_host(host)
-
- return "{}://{}:{}{}".format(self.__protocol, host,
- self.__http_server.port, suffix)
-
- def url_base(self, host='localhost') -> str:
+ def url_base(self) -> str:
"""Returns the url for the base page to navigate and prevent CORS.
"""
- return self._url_for(self.__path_base, host)
+ return self.__http_server.url_for(self.__path_base)
- def url_200(self, host='localhost') -> str:
+ def url_200(self, content=None) -> str:
"""Returns the url for the 200 page with the `default_200_page_content`.
"""
- return self._url_for(self.__path_200, host)
+ if content is not None:
+ path = f"{self.__path_200}/{str(uuid.uuid4())}"
+ self.__http_server \
+ .expect_request(path) \
+ .respond_with_data(
+ self.__html_doc(content),
+ headers={"Content-Type": "text/html"})
+
+ return self.__http_server.url_for(path)
+
+ return self.__http_server.url_for(self.__path_200)
def url_permanent_redirect(self) -> str:
"""Returns the url for the permanent redirect page, redirecting to the
200 page."""
- return self._url_for(self.__path_permanent_redirect)
+ return self.__http_server.url_for(self.__path_permanent_redirect)
def url_basic_auth(self) -> str:
"""Returns the url for the page with a basic auth."""
- return self._url_for(self.__path_basic_auth)
+ return self.__http_server.url_for(self.__path_basic_auth)
def url_hang_forever(self) -> str:
"""Returns the url for the page, request to which will never be finished."""
- return self._url_for(self.__path_hang_forever)
+ return self.__http_server.url_for(self.__path_hang_forever)
- def url_cacheable(self, host='localhost') -> str:
+ def url_cacheable(self) -> str:
"""Returns the url for the cacheable page with the `default_200_page_content`."""
- return self._url_for(self.__path_cacheable, host)
+ return self.__http_server.url_for(self.__path_cacheable)
diff --git a/tests/tools/test_local_http_server.py b/tests/tools/test_local_http_server.py
index df06e3110c..4876950535 100644
--- a/tests/tools/test_local_http_server.py
+++ b/tests/tools/test_local_http_server.py
@@ -50,6 +50,14 @@ async def test_local_server_200(websocket, context_id, local_server_http):
== local_server_http.content_200
+@pytest.mark.asyncio
+async def test_local_server_custom_content(websocket, context_id,
+ local_server_http):
+ some_custom_content = 'some custom content'
+ assert await get_content(websocket, context_id, local_server_http.url_200(content=some_custom_content)) \
+ == some_custom_content
+
+
@pytest.mark.asyncio
async def test_local_server_redirect(websocket, context_id, local_server_http):
assert await get_content(websocket, context_id,
diff --git a/tools/run_local_http_server.py b/tools/run_local_http_server.py
index 4e4351e996..f04612a5b1 100644
--- a/tools/run_local_http_server.py
+++ b/tools/run_local_http_server.py
@@ -24,10 +24,13 @@
import local_http_server # noqa: E402
local_server_http = local_http_server.LocalHttpServer()
+local_server_http_another_origin = local_http_server.LocalHttpServer(
+ host='127.0.0.1')
local_server_bad_ssl = local_http_server.LocalHttpServer(protocol='https')
print(f"""Local http server started...
- 200: {local_server_http.url_200()}
+ - oopif: {local_server_http.url_200(content='')}
- 301 / permanent redirect: {local_server_http.url_permanent_redirect()}
- 401 / basic auth: {local_server_http.url_basic_auth()}
- hangs forever: {local_server_http.url_hang_forever()}