Make free_host_port a fixture

tests/by_image/base-notebook/test_container_options.py

@@ -6,20 +6,20 @@
 import pytest  # type: ignore
 import requests
 
-from tests.utils.find_free_port import find_free_port
 from tests.utils.tracked_container import TrackedContainer
 
 LOGGER = logging.getLogger(__name__)
 
 
-def test_cli_args(container: TrackedContainer, http_client: requests.Session) -> None:
+def test_cli_args(
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
+) -> None:
     """Image should respect command line args (e.g., disabling token security)"""
-    host_port = find_free_port()
     container.run_detached(
         command=["start-notebook.py", "--IdentityProvider.token=''"],
-        ports={"8888/tcp": host_port},
+        ports={"8888/tcp": free_host_port},
     )
-    resp = http_client.get(f"http://localhost:{host_port}")
+    resp = http_client.get(f"http://localhost:{free_host_port}")
     resp.raise_for_status()
     logs = container.get_logs()
     LOGGER.debug(logs)
@@ -54,22 +54,21 @@ def test_nb_user_change(container: TrackedContainer) -> None:
 
 @pytest.mark.filterwarnings("ignore:Unverified HTTPS request")
 def test_unsigned_ssl(
-    container: TrackedContainer, http_client: requests.Session
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
 ) -> None:
     """Container should generate a self-signed SSL certificate
     and Jupyter Server should use it to enable HTTPS.
     """
-    host_port = find_free_port()
     container.run_detached(
         environment=["GEN_CERT=yes"],
-        ports={"8888/tcp": host_port},
+        ports={"8888/tcp": free_host_port},
     )
     # NOTE: The requests.Session backing the http_client fixture
     # does not retry properly while the server is booting up.
     # An SSL handshake error seems to abort the retry logic.
     # Forcing a long sleep for the moment until I have time to dig more.
     time.sleep(1)
-    resp = http_client.get(f"https://localhost:{host_port}", verify=False)
+    resp = http_client.get(f"https://localhost:{free_host_port}", verify=False)
     resp.raise_for_status()
     assert "login_submit" in resp.text
     logs = container.get_logs()
@@ -94,18 +93,18 @@ def test_unsigned_ssl(
 def test_custom_internal_port(
     container: TrackedContainer,
     http_client: requests.Session,
+    free_host_port: int,
     env: dict[str, str],
 ) -> None:
     """Container should be accessible from the host
     when using custom internal port"""
-    host_port = find_free_port()
     internal_port = env.get("JUPYTER_PORT", 8888)
     container.run_detached(
         command=["start-notebook.py", "--IdentityProvider.token=''"],
         environment=env,
-        ports={internal_port: host_port},
+        ports={internal_port: free_host_port},
     )
-    resp = http_client.get(f"http://localhost:{host_port}")
+    resp = http_client.get(f"http://localhost:{free_host_port}")
     resp.raise_for_status()
     logs = container.get_logs()
     LOGGER.debug(logs)

tests/by_image/base-notebook/test_notebook.py

@@ -2,16 +2,14 @@
 # Distributed under the terms of the Modified BSD License.
 import requests
 
-from tests.utils.find_free_port import find_free_port
 from tests.utils.tracked_container import TrackedContainer
 
 
 def test_secured_server(
-    container: TrackedContainer, http_client: requests.Session
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
 ) -> None:
     """Jupyter Server should eventually request user login."""
-    host_port = find_free_port()
-    container.run_detached(ports={"8888/tcp": host_port})
-    resp = http_client.get(f"http://localhost:{host_port}")
+    container.run_detached(ports={"8888/tcp": free_host_port})
+    resp = http_client.get(f"http://localhost:{free_host_port}")
     resp.raise_for_status()
     assert "login_submit" in resp.text, "User login not requested"

tests/by_image/base-notebook/test_start_container.py

@@ -6,7 +6,6 @@
 import pytest  # type: ignore
 import requests
 
-from tests.utils.find_free_port import find_free_port
 from tests.utils.tracked_container import TrackedContainer
 
 LOGGER = logging.getLogger(__name__)
@@ -32,6 +31,7 @@
 def test_start_notebook(
     container: TrackedContainer,
     http_client: requests.Session,
+    free_host_port: int,
     env: list[str] | None,
     expected_command: str,
     expected_start: bool,
@@ -41,8 +41,7 @@ def test_start_notebook(
     LOGGER.info(
         f"Test that the start-notebook.py launches the {expected_command} server from the env {env} ..."
     )
-    host_port = find_free_port()
-    container.run_detached(environment=env, ports={"8888/tcp": host_port})
+    container.run_detached(environment=env, ports={"8888/tcp": free_host_port})
     # sleeping some time to let the server start
     time.sleep(2)
     logs = container.get_logs()
@@ -59,7 +58,7 @@ def test_start_notebook(
     assert len(expected_warnings) == len(warnings)
     # checking if the server is listening
     if expected_start:
-        resp = http_client.get(f"http://localhost:{host_port}")
+        resp = http_client.get(f"http://localhost:{free_host_port}")
         assert resp.status_code == 200, "Server is not listening"
 
 

tests/by_image/datascience-notebook/test_pluto_datascience.py

@@ -7,7 +7,7 @@
 
 
 def test_pluto_proxy(
-    container: TrackedContainer, http_client: requests.Session
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
 ) -> None:
     """Pluto proxy starts Pluto correctly"""
-    check_pluto_proxy(container, http_client)
+    check_pluto_proxy(container, http_client, free_host_port)

tests/by_image/julia-notebook/test_pluto.py

@@ -7,7 +7,7 @@
 
 
 def test_pluto_proxy(
-    container: TrackedContainer, http_client: requests.Session
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
 ) -> None:
     """Pluto proxy starts Pluto correctly"""
-    check_pluto_proxy(container, http_client)
+    check_pluto_proxy(container, http_client, free_host_port)

tests/conftest.py

@@ -2,7 +2,9 @@
 # Distributed under the terms of the Modified BSD License.
 import logging
 import os
+import socket
 from collections.abc import Generator
+from contextlib import closing
 
 import docker
 import pytest  # type: ignore
@@ -54,3 +56,12 @@ def container(
     )
     yield container
     container.remove()
+
+
+@pytest.fixture(scope="function")
+def free_host_port() -> Generator[int]:
+    """Finds a free port on the host machine"""
+    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as s:
+        s.bind(("", 0))
+        s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        yield s.getsockname()[1]

tests/shared_checks/pluto_check.py

@@ -6,26 +6,24 @@
 
 import requests
 
-from tests.utils.find_free_port import find_free_port
 from tests.utils.tracked_container import TrackedContainer
 
 LOGGER = logging.getLogger(__name__)
 
 
 def check_pluto_proxy(
-    container: TrackedContainer, http_client: requests.Session
+    container: TrackedContainer, http_client: requests.Session, free_host_port: int
 ) -> None:
-    host_port = find_free_port()
     token = secrets.token_hex()
     container.run_detached(
         command=[
             "start-notebook.py",
             f"--IdentityProvider.token={token}",
         ],
-        ports={"8888/tcp": host_port},
+        ports={"8888/tcp": free_host_port},
     )
     # Give the server a bit of time to start
     time.sleep(2)
-    resp = http_client.get(f"http://localhost:{host_port}/pluto?token={token}")
+    resp = http_client.get(f"http://localhost:{free_host_port}/pluto?token={token}")
     resp.raise_for_status()
     assert "Pluto.jl notebooks" in resp.text, "Pluto.jl text not found in /pluto page"