Test server listening on IPv4/IPv6

tests/base-notebook/data/check_listening.py

@@ -0,0 +1,51 @@
+#!/usr/bin/env python
+# Copyright (c) Jupyter Development Team.
+# Distributed under the terms of the Modified BSD License.
+import socket
+import time
+
+import requests
+
+
+def test_connect() -> None:
+    # Give some time for server to start
+    finish_time = time.time() + 10
+    sleep_time = 1
+    while time.time() < finish_time:
+        time.sleep(sleep_time)
+        try:
+            requests.get("http://localhost:8888/api")
+            break
+        except requests.RequestException:
+            pass
+
+    # https://docs.python.org/3/library/socket.html#socket.getaddrinfo
+    ipv4_addrs = {
+        s[4][0] for s in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET)
+    }
+    ipv4_addrs.discard("127.0.0.1")
+    if len(ipv4_addrs) < 1:
+        raise Exception("No external IPv4 addresses found")
+    for addr in ipv4_addrs:
+        url = f"http://{addr}:8888/api"
+        r = requests.get(url)
+        r.raise_for_status()
+        assert "version" in r.json()
+        print(f"Successfully connected to IPv4 {url}")
+
+    ipv6_addrs = {
+        s[4][0] for s in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET6)
+    }
+    ipv6_addrs.discard("::1")
+    if len(ipv6_addrs) < 1:
+        raise Exception("No external IPv6 addresses found")
+    for addr in ipv6_addrs:
+        url = f"http://[{addr}]:8888/api"
+        r = requests.get(url)
+        r.raise_for_status()
+        assert "version" in r.json()
+        print(f"Successfully connected to IPv6 {url}")
+
+
+if __name__ == "__main__":
+    test_connect()

tests/base-notebook/test_ips.py

@@ -0,0 +1,40 @@
+# Copyright (c) Jupyter Development Team.
+# Distributed under the terms of the Modified BSD License.
+import logging
+import time
+from pathlib import Path
+
+from docker.models import Container
+
+from tests.conftest import TrackedContainer, get_health
+
+LOGGER = logging.getLogger(__name__)
+THIS_DIR = Path(__file__).parent.resolve()
+
+
+def wait_healthy(container: Container, timeout: int) -> None:
+    finish_time = time.time() + timeout
+    sleep_time = 1
+    while time.time() < finish_time:
+        time.sleep(sleep_time)
+        if get_health(container) == "healthy":
+            return
+
+    raise Exception(f"Container {container.name} not healthy after {int} s")
+
+
+def test_ipv46(container: TrackedContainer, ipv6_network: str) -> None:
+    """Check server is listening on the expected IP families"""
+    host_data_dir = THIS_DIR / "data"
+    cont_data_dir = "/home/jovyan/data"
+    LOGGER.info("Testing that server is listening on IPv4 and IPv6 ...")
+    running_container = container.run_detached(
+        network=ipv6_network,
+        volumes={str(host_data_dir): {"bind": cont_data_dir, "mode": "ro,z"}},
+        tty=True,
+    )
+
+    command = ["python", "./data/check_listening.py"]
+    r = running_container.exec_run(command)
+    LOGGER.info(r.output.decode())
+    assert r.exit_code == 0

tests/conftest.py

@@ -3,8 +3,10 @@
 import logging
 import os
 import socket
+from collections.abc import Generator
 from contextlib import closing
 from typing import Any, Optional
+from uuid import uuid4
 
 import docker
 import pytest  # type: ignore
@@ -52,6 +54,15 @@ def image_name() -> str:
     return os.environ["TEST_IMAGE"]
 
 
+@pytest.fixture(scope="session")
+def ipv6_network(docker_client: docker.DockerClient) -> Generator[str, None, None]:
+    """Create a dual-stack IPv6 docker network"""
+    name = str(uuid4())
+    docker_client.networks.create(name, enable_ipv6=True)
+    yield name
+    docker_client.networks.get(name).remove()
+
+
 class TrackedContainer:
     """Wrapper that collects docker container configuration and delays
     container creation/execution.