diff --git a/tests/browsing_context/test_download.py b/tests/browsing_context/test_download.py index 7b644aa29c..232a3346ad 100644 --- a/tests/browsing_context/test_download.py +++ b/tests/browsing_context/test_download.py @@ -147,10 +147,10 @@ async def test_browsing_context_download_finished_complete( @pytest.mark.asyncio async def test_browsing_context_download_finished_canceled( websocket, test_headless_mode, url_hang_forever_download, context_id, - html, get_cdp_session_id, filename): + html, get_cdp_session_id, filename, some_host): page_url = html( - f"""Download""" + f"""Download""" ) await goto_url(websocket, context_id, page_url) @@ -215,7 +215,7 @@ async def test_browsing_context_download_finished_canceled( 'navigation': navigation_id, 'status': 'canceled', 'timestamp': ANY_TIMESTAMP, - 'url': url_hang_forever_download(), + 'url': url_hang_forever_download, }, 'type': 'event', } diff --git a/tests/browsing_context/test_navigate.py b/tests/browsing_context/test_navigate.py index 047739bdea..803cb99ef3 100644 --- a/tests/browsing_context/test_navigate.py +++ b/tests/browsing_context/test_navigate.py @@ -917,7 +917,8 @@ async def test_speculationrules_disable_prerender(websocket, context_id, html): @pytest.mark.asyncio async def test_browsing_context_navigate_ssl_bad(websocket, context_id, - local_server_bad_ssl): + local_server_bad_ssl, + some_host): with pytest.raises(Exception, match=str({ "error": "unknown error", @@ -927,7 +928,7 @@ async def test_browsing_context_navigate_ssl_bad(websocket, context_id, websocket, { 'method': "browsingContext.navigate", 'params': { - 'url': local_server_bad_ssl.url_200(), + 'url': local_server_bad_ssl.url_200(host=some_host), 'wait': 'complete', 'context': context_id } @@ -936,12 +937,13 @@ async def test_browsing_context_navigate_ssl_bad(websocket, context_id, @pytest.mark.asyncio async def test_browsing_context_navigate_ssl_good(websocket, context_id, - local_server_good_ssl): + local_server_good_ssl, + some_host): await execute_command( websocket, { 'method': "browsingContext.navigate", 'params': { - 'url': local_server_good_ssl.url_200(), + 'url': local_server_good_ssl.url_200(host=some_host), 'wait': 'complete', 'context': context_id } diff --git a/tests/conftest.py b/tests/conftest.py index e48735b247..20f00145e6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -38,25 +38,21 @@ @pytest_asyncio.fixture(scope='session') -def local_server_http() -> Generator[LocalHttpServer, None, None]: - """ - Returns an instance of a LocalHttpServer without SSL pointing to localhost. - """ - server = LocalHttpServer() - yield server +def some_host() -> str: + return "localhost" - server.clear() - if server.is_running(): - server.stop() - return + +@pytest_asyncio.fixture(scope='session') +def another_host() -> str: + return "another_host.test" @pytest_asyncio.fixture(scope='session') -def local_server_http_another_host() -> Generator[LocalHttpServer, None, None]: +def local_server_http() -> Generator[LocalHttpServer, None, None]: """ - Returns an instance of a LocalHttpServer without SSL pointing to `127.0.0.1` + Returns an instance of a LocalHttpServer without SSL pointing to localhost. """ - server = LocalHttpServer('127.0.0.1') + server = LocalHttpServer() yield server server.clear() @@ -138,7 +134,9 @@ async def create_session(connection): # Required to prevent automatic switch to https. "--disable-features=HttpsFirstBalancedModeAutoEnable,HttpsUpgrades,LocalNetworkAccessChecks", # Required for bluetooth testing. - "--enable-features=WebBluetooth" + "--enable-features=WebBluetooth", + # Required for testing with `.test` domains. + "--host-resolver-rules=MAP *.test 127.0.0.1", ] } } @@ -341,46 +339,46 @@ def url_all_origins(request, url_example, url_example_another_origin, html): @pytest.fixture -def url_base(local_server_http): +def url_base(local_server_http, some_host): """Return a generic example URL with status code 200.""" - return local_server_http.url_base() + return local_server_http.url_base(host=some_host) @pytest.fixture -def url_example(local_server_http): +def url_example(local_server_http, some_host): """Return a generic example URL with status code 200.""" - return local_server_http.url_200() + return local_server_http.url_200(host=some_host) @pytest.fixture -def url_example_another_origin(local_server_http_another_host): +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_another_host.url_200() + return local_server_http.url_200(host=another_host) @pytest.fixture -def url_auth_required(local_server_http): +def url_auth_required(local_server_http, some_host): """Return a URL that requires authentication (status code 401). Alternatively, any of the following URLs could also be used: - "https://authenticationtest.com/HTTPAuth/" - "http://the-internet.herokuapp.com/basic_auth" - "http://httpstat.us/401" """ - return local_server_http.url_basic_auth() + return local_server_http.url_basic_auth(host=some_host) @pytest.fixture -def url_hang_forever(local_server_http): +def url_hang_forever(local_server_http, some_host): """Return a URL that hangs forever.""" try: - yield local_server_http.url_hang_forever() + yield local_server_http.url_hang_forever(host=some_host) finally: local_server_http.hang_forever_stop() @pytest.fixture(scope="session") -def url_bad_ssl(local_server_bad_ssl): +def url_bad_ssl(local_server_bad_ssl, some_host): """ Return a URL with an invalid certificate authority from a SSL certificate. In Chromium, this generates the following error: @@ -388,18 +386,18 @@ def url_bad_ssl(local_server_bad_ssl): > Your connection is not private > NET::ERR_CERT_AUTHORITY_INVALID """ - return local_server_bad_ssl.url_200() + return local_server_bad_ssl.url_200(host=some_host) @pytest.fixture(scope="session") -def url_secure_context(local_server_good_ssl): - return local_server_good_ssl.url_200() +def url_secure_context(local_server_good_ssl, some_host): + return local_server_good_ssl.url_200(host=some_host) @pytest.fixture -def url_cacheable(local_server_http): +def url_cacheable(local_server_http, some_host): """Return a generic example URL that can be cached.""" - return local_server_http.url_cacheable() + return local_server_http.url_cacheable(host=some_host) @pytest.fixture @@ -588,11 +586,12 @@ async def activate_main_tab(): @pytest.fixture -def url_download(local_server_http): +def url_download(local_server_http, some_host): """Return a URL that triggers a download.""" def url_download(file_name="file-name.txt", content="download content"): return local_server_http.url_200( - content, + host=some_host, + content=content, content_type="application/octet-stream", headers={ "Content-Disposition": f"attachment; filename=\"{file_name}\"" @@ -602,22 +601,20 @@ def url_download(file_name="file-name.txt", content="download content"): @pytest.fixture -def url_hang_forever_download(local_server_http): +def url_hang_forever_download(local_server_http, some_host): """Return a URL that triggers a download which hangs forever.""" try: - yield local_server_http.url_hang_forever_download + yield local_server_http.url_hang_forever_download(host=some_host) finally: local_server_http.hang_forever_stop() @pytest.fixture -def html(local_server_http, local_server_http_another_host): +def html(local_server_http, some_host, another_host): """Return a factory for URL with the given content.""" def html(content="", same_origin=True): - if same_origin: - return local_server_http.url_200(content=content) - else: - return local_server_http_another_host.url_200(content=content) + return local_server_http.url_200( + host=(some_host if same_origin else another_host), content=content) return html diff --git a/tests/emulation/test_geolocation.py b/tests/emulation/test_geolocation.py index 8ab646bd16..b80d418f98 100644 --- a/tests/emulation/test_geolocation.py +++ b/tests/emulation/test_geolocation.py @@ -166,15 +166,12 @@ async def test_geolocation_emulate_unavailable(websocket, context_id, @pytest.mark.asyncio async def test_geolocation_per_user_context(websocket, url_example, - url_example_another_origin, + url_secure_context, user_context_id, create_context, snapshot): - # `url_example_another_origin` is required as `local_server_http` can - # be dead-locked in case of concurrent requests. - await set_permission(websocket, get_origin(url_example), {'name': 'geolocation'}, 'granted', "default") - await set_permission(websocket, get_origin(url_example_another_origin), + await set_permission(websocket, get_origin(url_secure_context), {'name': 'geolocation'}, 'granted', user_context_id) # Set different geolocation overrides for different user contexts. @@ -203,8 +200,7 @@ async def test_geolocation_per_user_context(websocket, url_example, assert emulated_geolocation_1 == snapshot() browsing_context_id_2 = await create_context(user_context_id) - await goto_url(websocket, browsing_context_id_2, - url_example_another_origin) + await goto_url(websocket, browsing_context_id_2, url_secure_context) emulated_geolocation_2 = await get_geolocation(websocket, browsing_context_id_2) assert emulated_geolocation_2 == snapshot() diff --git a/tests/script/test_realm.py b/tests/script/test_realm.py index 94eb8480d9..cc2368dcbe 100644 --- a/tests/script/test_realm.py +++ b/tests/script/test_realm.py @@ -15,7 +15,7 @@ import pytest from anys import ANY_DICT, ANY_NUMBER, ANY_STR -from test_helpers import (AnyExtending, execute_command, goto_url, +from test_helpers import (AnyExtending, execute_command, get_origin, goto_url, read_JSON_message, send_JSON_command, subscribe, wait_for_event, wait_for_filtered_event) @@ -44,7 +44,7 @@ async def test_realm_realmCreated(websocket, context_id, html, "method": "script.realmCreated", "params": { "type": "window", - "origin": local_server_http.origin(), + "origin": get_origin(url), "realm": ANY_STR, "context": context_id, } diff --git a/tests/service_worker/test_navigate.py b/tests/service_worker/test_navigate.py index a3b98535f9..3d24fcaa35 100644 --- a/tests/service_worker/test_navigate.py +++ b/tests/service_worker/test_navigate.py @@ -24,10 +24,11 @@ }], indirect=True) async def test_serviceWorker_acceptInsecureCertsCapability_respected( - websocket, context_id, local_server_bad_ssl): + websocket, context_id, local_server_bad_ssl, some_host): service_worker_script = local_server_bad_ssl.url_200( - content='', content_type='text/javascript') - service_worker_page = local_server_bad_ssl.url_200(content=f"""""") diff --git a/tests/tools/local_http_server.py b/tests/tools/local_http_server.py index 4af7fbd3ee..b9d4425d68 100644 --- a/tests/tools/local_http_server.py +++ b/tests/tools/local_http_server.py @@ -29,6 +29,9 @@ from flask import Response as FlaskResponse from flask import redirect, request, stream_with_context +LOCAL_HOST = 'localhost' +LOCAL_IP = '127.0.0.1' + # Helper to find a free port def find_free_port() -> int: @@ -96,11 +99,10 @@ def stop(self) -> None: def __html_doc(self, content: str) -> str: return f"{content}" - def __init__(self, host: str = 'localhost', ssl_cert_prefix=None) -> None: + def __init__(self, ssl_cert_prefix=None) -> None: self.__app = Flask(__name__) # Important for some Flask behaviors in a test context self.__app.testing = True - self.__host = host self.__protocol = 'http' if ssl_cert_prefix is None else "https" self.__port = find_free_port() @@ -152,7 +154,7 @@ def route_200_dynamic(response_id: str): @self.__app.route(self.__path_permanent_redirect) def route_permanent_redirect(): - return redirect(self.url_200(), code=301) + return redirect(self.url_200(host=LOCAL_IP), code=301) @self.__app.route(self.__path_basic_auth) def process_auth(): @@ -238,12 +240,12 @@ def _check_server_readiness(self, timeout_s: float = 0.1) -> bool: context.check_hostname = False # For self-signed certs context.verify_mode = ssl.CERT_NONE - conn = http.client.HTTPSConnection(self.__host, + conn = http.client.HTTPSConnection(LOCAL_IP, self.__port, timeout=timeout_s, context=context) else: - conn = http.client.HTTPConnection(self.__host, + conn = http.client.HTTPConnection(LOCAL_IP, self.__port, timeout=timeout_s) @@ -270,7 +272,7 @@ def _wait_for_server_startup(self, max_wait_s: int = 5) -> None: # Short sleep before retrying time.sleep(0.05) raise RuntimeError( - f"Flask server failed to start on {self.__protocol}://{self.__host}:{self.__port} within {max_wait_s}s." + f"Flask server failed to start on port {self.__port} within {max_wait_s}s." ) def _start_server(self, @@ -280,7 +282,7 @@ def _start_server(self, return kwargs = { - 'host': self.__host, + 'host': LOCAL_IP, 'port': self.__port, # Should be False for stability and threaded mode 'debug': False, @@ -306,19 +308,16 @@ def hang_forever_stop(self): # Release any hanging requests self.hang_forever_stop_flag.set() - def _build_url(self, path: str) -> str: + def _build_url(self, host: str, path: str) -> str: """Constructs a full URL for a given path on this server.""" - return f"{self.__protocol}://{self.__host}:{self.__port}{path}" - - def origin(self) -> str: - """Returns the origin (scheme://host:port) of the server.""" - return f"{self.__protocol}://{self.__host}:{self.__port}" + return f"{self.__protocol}://{host}:{self.__port}{path}" - def url_base(self) -> str: + def url_base(self, host) -> str: """Returns the URL for the base page (used to prevent CORS issues).""" - return self._build_url(self.__path_base) + return self._build_url(host, self.__path_base) def url_200(self, + host, content: str | None = None, content_type: str = "text/html", headers: dict[str, str] | None = None) -> str: @@ -345,26 +344,26 @@ def url_200(self, "headers": headers } path = f"{self.__path_200}/{response_id}" - return self._build_url(path) + return self._build_url(host, path) - return self._build_url(self.__path_200) + return self._build_url(host, self.__path_200) - def url_permanent_redirect(self) -> str: + def url_permanent_redirect(self, host) -> str: """Returns the URL for a page that permanently redirects to the default 200 page.""" - return self._build_url(self.__path_permanent_redirect) + return self._build_url(host, self.__path_permanent_redirect) - def url_basic_auth(self) -> str: + def url_basic_auth(self, host) -> str: """Returns the URL for a page protected by Basic authentication.""" - return self._build_url(self.__path_basic_auth) + return self._build_url(host, self.__path_basic_auth) - def url_hang_forever(self) -> str: + def url_hang_forever(self, host) -> str: """Returns the URL for a page that will hang until `hang_forever_stop()` is called.""" - return self._build_url(self.__path_hang_forever) + return self._build_url(host, self.__path_hang_forever) - def url_hang_forever_download(self) -> str: + def url_hang_forever_download(self, host) -> str: """Returns the URL for a page that will hang until `hang_forever_stop()` is called.""" - return self._build_url(self.__path_hang_forever_download) + return self._build_url(host, self.__path_hang_forever_download) - def url_cacheable(self) -> str: + def url_cacheable(self, host) -> str: """Returns the URL for a cacheable page (using Last-Modified and If-Modified-Since).""" - return self._build_url(self.__path_cacheable) + return self._build_url(host, self.__path_cacheable) diff --git a/tests/tools/test_local_http_server.py b/tests/tools/test_local_http_server.py index 77dab3f8a9..170110bb16 100644 --- a/tests/tools/test_local_http_server.py +++ b/tests/tools/test_local_http_server.py @@ -45,30 +45,28 @@ async def get_content(websocket, context_id, url): @pytest.mark.asyncio -async def test_local_server_200(websocket, context_id, local_server_http): - assert await get_content(websocket, context_id, local_server_http.url_200()) \ +async def test_local_server_200(websocket, context_id, local_server_http, + some_host): + assert await get_content(websocket, context_id, local_server_http.url_200(host=some_host)) \ == local_server_http.content_200 -@pytest.mark.asyncio -async def test_local_server_http_another_host_200( - websocket, context_id, local_server_http_another_host): - assert await get_content(websocket, context_id, local_server_http_another_host.url_200()) \ - == local_server_http_another_host.content_200 - - @pytest.mark.asyncio async def test_local_server_custom_content(websocket, context_id, - local_server_http): + local_server_http, some_host): 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 + assert await get_content( + websocket, context_id, + local_server_http.url_200( + host=some_host, + content=some_custom_content)) == some_custom_content @pytest.mark.asyncio -async def test_local_server_redirect(websocket, context_id, local_server_http): +async def test_local_server_redirect(websocket, context_id, local_server_http, + some_host): assert await get_content(websocket, context_id, - local_server_http.url_permanent_redirect()) \ + local_server_http.url_permanent_redirect(some_host)) \ == local_server_http.content_200 diff --git a/tools/run_local_http_server.py b/tools/run_local_http_server.py index 08d1af822b..76d8c3e42b 100644 --- a/tools/run_local_http_server.py +++ b/tools/run_local_http_server.py @@ -25,17 +25,16 @@ 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') +local_server_bad_ssl = local_http_server.LocalHttpServer( + ssl_cert_prefix="ssl_bad") 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()} - - bad ssl: {local_server_bad_ssl.url_200()} + - 200: {local_server_http.url_200('localhost')} + - oopif: {local_server_http.url_200('localhost', content='')} + - 301 / permanent redirect: {local_server_http.url_permanent_redirect('localhost')} + - 401 / basic auth: {local_server_http.url_basic_auth('localhost')} + - hangs forever: {local_server_http.url_hang_forever('localhost')} + - bad ssl: {local_server_bad_ssl.url_200('localhost')} """) while True: