test: test on *.test domains

Instead of testing on `localhost`, provide `--host-resolver-rules` arg to Chrome and test on `*.test` domains.

Author: sadym-chromium

tests/conftest.py

@@ -30,26 +30,27 @@
 from tools.local_http_server import LocalHttpServer
 
 
-@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
+@pytest.fixture(scope="session")
+def some_domain():
+    return 'some_domain.test'
 
-    server.clear()
-    if server.is_running():
-        server.stop()
-        return
+
+@pytest.fixture(scope="session")
+def another_domain():
+    return 'another_domain.test'
+
+
+@pytest.fixture(scope="session")
+def ssl_domain():
+    return 'ssl_domain.test'
 
 
 @pytest_asyncio.fixture(scope='session')
-def local_server_http_another_host() -> Generator[LocalHttpServer, None, None]:
+def local_http_server(some_domain) -> 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(default_host=some_domain)
     yield server
 
     server.clear()
@@ -59,9 +60,9 @@ def local_server_http_another_host() -> Generator[LocalHttpServer, None, None]:
 
 
 @pytest_asyncio.fixture(scope='session')
-def local_server_bad_ssl() -> Generator[LocalHttpServer, None, None]:
+def local_server_bad_ssl(ssl_domain) -> Generator[LocalHttpServer, None, None]:
     """ Returns an instance of a LocalHttpServer with bad SSL certificate. """
-    server = LocalHttpServer(protocol='https')
+    server = LocalHttpServer(default_host=ssl_domain, protocol='https')
     yield server
 
     server.clear()
@@ -116,22 +117,28 @@ async def capabilities(request):
 async def websocket(_websocket_connection, test_headless_mode, capabilities,
                     request):
     """Return a websocket with an active BiDi session."""
-    default_capabilities = {"webSocketUrl": True, "goog:chromeOptions": {}}
+    default_capabilities = {
+        "webSocketUrl": True,
+        "goog:chromeOptions": {
+            "args": [
+                # Required for testing with `.test` domains.
+                '--host-resolver-rules=MAP *.test 127.0.0.1',
+            ]
+        }
+    }
     maybe_browser_bin = os.getenv("BROWSER_BIN")
     if maybe_browser_bin:
         default_capabilities["goog:chromeOptions"][
             "binary"] = maybe_browser_bin
 
     if test_headless_mode != "false":
         if test_headless_mode == "old":
-            default_capabilities["goog:chromeOptions"]["args"] = [
-                "--headless=old", '--hide-scrollbars', '--mute-audio'
-            ]
+            default_capabilities["goog:chromeOptions"]["args"].extend(
+                ["--headless=old", "--hide-scrollbars", "--mute-audio"])
         else:
             # Default to new headless mode.
-            default_capabilities["goog:chromeOptions"]["args"] = [
-                "--headless=new"
-            ]
+            default_capabilities["goog:chromeOptions"]["args"].extend(
+                ["--headless=new"])
 
     session_capabilities = merge_dicts_recursively(default_capabilities,
                                                    capabilities)
@@ -267,60 +274,60 @@ def url_all_origins(request, url_example, url_example_another_origin, html):
 
 
 @pytest.fixture
-def url_base(local_server_http):
+def url_base(local_http_server):
     """Return a generic example URL with status code 200."""
-    return local_server_http.url_base()
+    return local_http_server.url_base()
 
 
 @pytest.fixture
-def url_example(local_server_http):
+def url_example(local_http_server):
     """Return a generic example URL with status code 200."""
-    return local_server_http.url_200()
+    return local_http_server.url_200()
 
 
 @pytest.fixture
-def url_example_another_origin(local_server_http_another_host):
+def url_example_another_origin(local_http_server, another_domain):
     """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_http_server.url_200(host=another_domain)
 
 
 @pytest.fixture
-def url_auth_required(local_server_http):
+def url_auth_required(local_http_server):
     """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_http_server.url_basic_auth()
 
 
 @pytest.fixture
-def url_hang_forever(local_server_http):
+def url_hang_forever(local_http_server):
     """Return a URL that hangs forever."""
     try:
-        yield local_server_http.url_hang_forever()
+        yield local_http_server.url_hang_forever()
     finally:
-        local_server_http.hang_forever_stop()
+        local_http_server.hang_forever_stop()
 
 
 @pytest.fixture(scope="session")
-def url_bad_ssl(local_server_bad_ssl):
+def url_bad_ssl(local_server_bad_ssl, ssl_domain):
     """
     Return a URL with an invalid certificate authority from a SSL certificate.
     In Chromium, this generates the following error:
 
     > 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=ssl_domain)
 
 
 @pytest.fixture
-def url_cacheable(local_server_http):
+def url_cacheable(local_http_server):
     """Return a generic example URL that can be cached."""
-    return local_server_http.url_cacheable()
+    return local_http_server.url_cacheable()
 
 
 @pytest.fixture
@@ -509,11 +516,11 @@ async def activate_main_tab():
 
 
 @pytest.fixture
-def url_download(local_server_http):
+def url_download(local_http_server):
     """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,
+        return local_http_server.url_200(
+            content=content,
             content_type="text/html",
             headers={
                 "Content-Disposition": f"attachment;  filename=\"{file_name}\""
@@ -523,13 +530,14 @@ def url_download(file_name="file-name.txt", content="download content"):
 
 
 @pytest.fixture
-def html(local_server_http, local_server_http_another_host):
+def html(local_http_server, some_domain, another_domain):
     """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)
+            return local_http_server.url_200(host=some_domain, content=content)
         else:
-            return local_server_http_another_host.url_200(content=content)
+            return local_http_server.url_200(host=another_domain,
+                                             content=content)
 
     return html
 

tests/network/test_network.py

@@ -148,7 +148,7 @@ async def test_network_global_subscription_enabled_in_new_context(
 
 @pytest.mark.asyncio
 async def test_network_before_request_sent_event_with_cookies_emitted(
-        websocket, context_id, url_base, url_example):
+        websocket, context_id, url_base, url_example, some_domain):
     pytest.xfail(
         "TODO: Fix flaky test https://github.com/GoogleChromeLabs/chromium-bidi/issues/2263"
     )
@@ -195,7 +195,7 @@ async def test_network_before_request_sent_event_with_cookies_emitted(
                 "headers": ANY_LIST,
                 "cookies": AnyOr([
                     AnyExtending({
-                        "domain": "localhost",
+                        "domain": some_domain,
                         "httpOnly": False,
                         "name": "foo",
                         "path": "/",

tests/script/test_realm.py

@@ -14,15 +14,15 @@
 # limitations under the License.
 
 import pytest
-from anys import ANY_NUMBER, ANY_STR
+from anys import ANY_NUMBER, ANY_STR, AnySubstr
 from test_helpers import (execute_command, read_JSON_message,
                           send_JSON_command, subscribe, wait_for_event,
                           wait_for_filtered_event)
 
 
 @pytest.mark.asyncio
 async def test_realm_realmCreated(websocket, context_id, html,
-                                  local_server_http):
+                                  local_http_server):
     url = html()
 
     await subscribe(websocket, ["script.realmCreated"])
@@ -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": AnySubstr(url),
             "realm": ANY_STR,
             "context": context_id,
         }

tests/tools/local_http_server.py

@@ -57,7 +57,7 @@ class LocalHttpServer:
     content_200: str = 'default 200 page'
 
     __app: Flask
-    __host: str
+    __default_host: str
     __port: int
     __protocol: Literal['http', 'https']
 
@@ -91,21 +91,21 @@ def stop(self) -> None:
         # Wait for the thread to terminate
         self._server_thread.join(timeout=5)
         if self._server_thread.is_alive():
-            print(
-                f"Warning: Flask server thread for {self.origin()} did not shut down cleanly after 5s."
-            )
+            print("Warning: Flask server did not shut down cleanly after 5s.")
         self._server_thread = None
 
     def __html_doc(self, content: str) -> str:
         return f"<!DOCTYPE html><html><head><link rel='shortcut icon' href='data:image/x-icon;,' type='image/x-icon'></head><body>{content}</body></html>"
 
-    def __init__(self,
-                 host: str = 'localhost',
-                 protocol: Literal['http', 'https'] = 'http') -> None:
+    def __init__(
+            self,
+            # Default host is used to resolve the URL, if a custom host is not provided.
+            default_host: str = 'localhost',
+            protocol: Literal['http', 'https'] = 'http') -> None:
         self.__app = Flask(__name__)
         # Important for some Flask behaviors in a test context
         self.__app.testing = True
-        self.__host = host
+        self.__default_host = default_host
         self.__protocol = protocol
         self.__port = find_free_port()
 
@@ -222,12 +222,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("localhost",
                                                    self.__port,
                                                    timeout=timeout_s,
                                                    context=context)
             else:
-                conn = http.client.HTTPConnection(self.__host,
+                conn = http.client.HTTPConnection("localhost",
                                                   self.__port,
                                                   timeout=timeout_s)
 
@@ -253,9 +253,7 @@ def _wait_for_server_startup(self, max_wait_s: int = 5) -> None:
                 return
             # 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."
-        )
+        raise RuntimeError("Flask server failed to start.")
 
     def _start_server(self,
                       ssl_context_config: tuple[str, str] | None) -> None:
@@ -264,7 +262,9 @@ def _start_server(self,
             return
 
         kwargs = {
-            'host': self.__host,
+            # Run server on local host.
+            'host': "localhost",
+            # Run server on the found free port.
             'port': self.__port,
             # Should be False for stability and threaded mode
             'debug': False,
@@ -290,19 +290,21 @@ def hang_forever_stop(self):
         # Release any hanging requests
         self.hang_forever_stop_flag.set()
 
-    def _build_url(self, path: str) -> str:
-        """Constructs a full URL for a given path on this server."""
-        return f"{self.__protocol}://{self.__host}:{self.__port}{path}"
+    def _build_url(self, host: str | None, path: str) -> str:
+        """Constructs a full URL for the given host and path on this server."""
+        return f"{self.__protocol}://{host if host is not None else self.__default_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}"
-
-    def url_base(self) -> str:
+    def url_base(
+        self,
+        host: str | None = None,
+    ) -> 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 if host is not None else self.__default_host,
+            self.__path_base)
 
     def url_200(self,
+                host: str | None = None,
                 content: str | None = None,
                 content_type: str = "text/html",
                 headers: dict[str, str] | None = None) -> str:
@@ -329,22 +331,22 @@ 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 | None = None) -> 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 | None = None) -> 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 | None = None) -> 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_cacheable(self) -> str:
+    def url_cacheable(self, host: str | None = None) -> 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)