Merge branch 'docker-port-publish' into 'main' (bbcfcbf1) · Commits · Dom Sekotill / Behave Utils

behave_utils/docker.py

+64 −3

Original line number	Diff line number	Diff line
		@@ -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,10 +320,18 @@ 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
		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):
		"""

tests/unit/docker/init.py

0 → 100644

+0 −0

Empty file added.

tests/unit/docker/test_misc.py

0 → 100644

+74 −0

Original line number	Diff line number	Diff line
		# Copyright 2022 Dominik Sekotill <dom.sekotill@kodo.org.uk>
		#
		# 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)