diff --git a/behave_utils/docker.py b/behave_utils/docker.py index 7f407683a636cca0652d5c0c53ba88e02f50b5cc..5c8ddbbac13c5849011c3346f0c667017ac74285 100644 --- a/behave_utils/docker.py +++ b/behave_utils/docker.py @@ -15,7 +15,9 @@ import ipaddress import json import logging from contextlib import contextmanager +from enum import Enum from os import PathLike +from os import environ from os import fspath from pathlib import Path from secrets import token_hex @@ -35,6 +37,7 @@ from typing import Tuple from typing import TypeVar from typing import Union from typing import cast +from urllib.parse import urlparse from .binaries import DownloadableDocker from .json import JSONArray @@ -47,12 +50,16 @@ from .proc import Executor from .proc import MutableArguments from .proc import exec_io +LOCALHOST = ipaddress.IPv4Address(0x7f000001) + MountPath = Union[PathLike[bytes], PathLike[str]] HostMount = tuple[MountPath, MountPath] NamedMount = tuple[str, MountPath] Mount = Union[HostMount, NamedMount, MountPath] Volumes = Iterable[Mount] +IPAddress = Union[ipaddress.IPv4Address, ipaddress.IPv6Address] + try: run([b"docker", b"version"], stdout=DEVNULL) @@ -91,6 +98,26 @@ def docker_quiet(*args: Argument, **env: str) -> None: run([DOCKER, *args], env=env, check=True, stdout=DEVNULL) +def _get_docker_host_ip() -> IPAddress: + """ + Return an IP address from the DOCKER_HOST environment variable, or a loopback address + + This function is *far* from complete, and there needs to be a much better way of + accessing ports on the Docker host. + + Currently, only IP addresses are supported, not names. + """ + url = environ.get("DOCKER_HOST") + if url is None: + return LOCALHOST + if not "://" in url: + url = f"tcp://{url}" + purl = urlparse(url) + if not purl.hostname or purl.scheme not in ("tcp", "ssh"): + return LOCALHOST + return ipaddress.ip_address(purl.hostname) + + class IPv4Address(ipaddress.IPv4Address): """ Subclass of IPv4Address that handle's docker idiosyncratic tendency to add a mask suffix @@ -107,6 +134,15 @@ class IPv4Address(ipaddress.IPv4Address): return cls(address) +class IPProtocol(Enum): + """ + IP protocols supported by Docker port forwarding + """ + + TCP = 'tcp' + UDP = 'udp' + + class Item: """ A mix-in for Docker items that can be inspected @@ -196,6 +232,7 @@ class Container(Item): network: Network|None = None, entrypoint: HostMount|Argument|None = None, privileged: bool = False, + publish: bool|list[int] = False, ): if isinstance(entrypoint, tuple): volumes = [*volumes, entrypoint] @@ -207,6 +244,7 @@ class Container(Item): self.env = env self.entrypoint = entrypoint self.privileged = privileged + self.publish = publish self.networks = dict[Network, Tuple[str, ...]]() self.cid: str|None = None @@ -260,8 +298,7 @@ class Container(Item): return self.cid opts: MutableArguments = [ - b"--network=none", - *(f"--env={name}={val}" for name, val in self.env.items()), + f"--env={name}={val}" for name, val in self.env.items() ] for vol in self.volumes: @@ -283,11 +320,19 @@ class Container(Item): if self.privileged: opts.append(b"--privileged") + if isinstance(self.publish, list): + opts.extend(f"--publish={p}" for p in self.publish) + elif self.publish: + opts.append(b"--publish-all") + else: + opts.append(b"--network=none") + self.cid = docker_output(b"container", b"create", *opts, self.image.iid, *self.cmd) assert self.cid # Disconnect the "none" network specified as the starting network - docker_quiet(b"network", b"disconnect", b"none", self.cid) + if not self.publish: + docker_quiet(b"network", b"disconnect", b"none", self.cid) return self.cid @@ -393,6 +438,22 @@ class Container(Item): stdin=stdin, stdout=stdout, stderr=stderr, ) + def get_external_ports(self, port: int, proto: IPProtocol = IPProtocol.TCP) -> Iterable[tuple[IPAddress, int]]: + """ + Yield (address, port) combinations exposed on the host that map to the given container port + """ + name = f"{port}/{proto.name.lower()}" + ports = self.inspect().path( + f"$.NetworkSettings.Ports.{name}", + list[dict[str, str]], + ) + if not ports: + raise KeyError(f"port {name} has not been published") + for portd in ports: + addr = ipaddress.ip_address(portd["HostIp"]) + port = int(portd["HostPort"]) + yield (_get_docker_host_ip() if addr.is_unspecified else addr), port + class Network(Item): """ diff --git a/tests/unit/docker/__init__.py b/tests/unit/docker/__init__.py new file mode 100644 index 0000000000000000000000000000000000000000..e69de29bb2d1d6434b8b29ae775ad8c2e48c5391 diff --git a/tests/unit/docker/test_misc.py b/tests/unit/docker/test_misc.py new file mode 100644 index 0000000000000000000000000000000000000000..99e183b11b450bb6fab17b7b6d629b7b91a6d182 --- /dev/null +++ b/tests/unit/docker/test_misc.py @@ -0,0 +1,74 @@ +# Copyright 2022 Dominik Sekotill +# +# This Source Code Form is subject to the terms of the Mozilla Public +# License, v. 2.0. If a copy of the MPL was not distributed with this +# file, You can obtain one at http://mozilla.org/MPL/2.0/. + +""" +Unit tests for miscellaneous functions of behave_utils.docker +""" + +from ipaddress import IPv4Address +from ipaddress import IPv6Address +from os import environ +from unittest.mock import patch + +from behave_utils import docker + +from .. import TestCase + + +@patch.dict(environ) +class HostIPTests(TestCase): + """ + Tests for the _get_docker_host_ip function + """ + + def test_missing(self) -> None: + """ + Check that a missing DOCKER_HOST returns a loopback address + """ + assert docker._get_docker_host_ip() == docker.LOCALHOST + + def test_non_ip(self) -> None: + """ + Check that a non-IP address in DOCKER_HOST returns a loopback address + """ + values = [ + "unix://foo/bar", + "fd://example.com", + ] + for environ["DOCKER_HOST"] in values: + with self.subTest(DOCKER_HOST=environ["DOCKER_HOST"]): + assert docker._get_docker_host_ip() == docker.LOCALHOST + assert docker._get_docker_host_ip() == docker.LOCALHOST + + def test_ipv4(self) -> None: + """ + Check that an IPv4 address in DOCKER_HOST returns that address + """ + addr = IPv4Address("10.42.0.1") + values = [ + f"{addr}", + f"tcp://{addr}", + f"tcp://{addr}:1234", + f"ssh://{addr}", + ] + for environ["DOCKER_HOST"] in values: + with self.subTest(DOCKER_HOST=environ["DOCKER_HOST"]): + assert docker._get_docker_host_ip() == addr + + def test_ipv6(self) -> None: + """ + Check that an IPv4 address in DOCKER_HOST returns that address + """ + addr = IPv6Address("2001:db8::0a2a:0001") + values = [ + f"tcp://[{addr}]", + f"tcp://[{addr}]:1234", + f"ssh://[{addr}]", + ] + for environ["DOCKER_HOST"] in values: + with self.subTest(DOCKER_HOST=environ["DOCKER_HOST"]): + # assert docker._get_docker_host_ip() == addr + self.assertEqual(docker._get_docker_host_ip(), addr)