diff --git a/tests/base-notebook/data/check_listening.py b/tests/base-notebook/data/check_listening.py new file mode 100644 index 0000000000..114ce5cfd2 --- /dev/null +++ b/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() diff --git a/tests/base-notebook/test_ips.py b/tests/base-notebook/test_ips.py new file mode 100644 index 0000000000..1ea6bcdf68 --- /dev/null +++ b/tests/base-notebook/test_ips.py @@ -0,0 +1,50 @@ +# Copyright (c) Jupyter Development Team. +# Distributed under the terms of the Modified BSD License. +import logging +from collections.abc import Generator +from pathlib import Path +from random import randint + +import docker +import pytest # type: ignore + +from tests.conftest import TrackedContainer + +LOGGER = logging.getLogger(__name__) +THIS_DIR = Path(__file__).parent.resolve() + + +@pytest.fixture(scope="session") +def ipv6_network(docker_client: docker.DockerClient) -> Generator[str, None, None]: + """Create a dual-stack IPv6 docker network""" + # Doesn't have to be routable since we're testing inside the container + subnet64 = "fc00:" + ":".join(hex(randint(0, 2**16))[2:] for _ in range(3)) + name = subnet64.replace(":", "-") + docker_client.networks.create( + name, + ipam=docker.types.IPAMPool( + subnet=subnet64 + "::/64", + gateway=subnet64 + "::1", + ), + enable_ipv6=True, + internal=True, + ) + yield name + docker_client.networks.get(name).remove() + + +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