Pre-assign addr when connecting Container->Network (a88f7010) · Commits · Dom Sekotill / Behave Utils

behave_utils/docker.py

+53 −7

Original line number	Diff line number	Diff line
		@@ -10,6 +10,7 @@ Commands for managing Docker for fixtures

		from __future__ import annotations

		import codecs
		import ipaddress
		import json
		import logging
		@@ -53,6 +54,13 @@ Volumes = Iterable[Mount]
		DOCKER = "docker"


		def utf8_decode(buffer: bytes) -> str:
		"""
		Return a decoded string from a bytes-like sequence of bytes
		"""
		return codecs.getdecoder("utf-8")(buffer)[0]


		def docker(args: Argument, *env: str) -> None:
		"""
		Run a Docker command, with output going to stdout
		@@ -277,7 +285,12 @@ class Container(Item):
		docker_quiet('container', 'rm', self.cid)
		self.cid = None

		def connect(self, network: Network, *aliases: str) -> None:
		def connect(
		self,
		network: Network,
		*aliases: str,
		address: ipaddress.IPv4Address\|ipaddress.IPv6Address\|None = None,
		) -> None:
		"""
		Connect the container to a Docker network

		@@ -285,15 +298,20 @@ class Container(Item):
		network.
		"""
		cid = self.get_id()
		opts = [f'--alias={a}' for a in aliases]

		if address is None:
		address = network.get_free_address()
		opts.append(
		f"--ip={address}" if isinstance(address, ipaddress.IPv4Address) else
		f"--ip6={address}",
		)

		if network in self.networks:
		if self.networks[network] == aliases:
		return
		docker('network', 'disconnect', str(network), cid)
		docker(
		'network', 'connect',
		*(f'--alias={a}' for a in aliases),
		str(network), cid,
		)
		docker('network', 'connect', *opts, str(network), cid)
		self.networks[network] = aliases

		def show_logs(self) -> None:
		@@ -353,6 +371,8 @@ class Network(Item):
		A Docker network
		"""

		DOCKER_SUBNET = ipaddress.IPv4Network("172.16.0.0/12")

		def __init__(self, name: str\|None = None) -> None:
		self._name = name or f"br{token_hex(6)}"

		@@ -395,7 +415,12 @@ class Network(Item):
		"""
		Create the network
		"""
		docker_quiet("network", "create", self._name)
		subnet = self.get_free_subnet()
		gateway = next(subnet.hosts())
		docker_quiet(
		"network", "create", self._name,
		f"--subnet={subnet}", f"--gateway={gateway}",
		)

		def destroy(self) -> None:
		"""
		@@ -403,6 +428,27 @@ class Network(Item):
		"""
		docker_quiet("network", "rm", self._name)

		@classmethod
		def get_free_subnet(cls) -> ipaddress.IPv4Network:
		"""
		Return a free private subnet
		"""
		networks = exec_io(
		[DOCKER, "network", "ls", "--format={{.ID}}"],
		deserialiser=utf8_decode,
		).splitlines()
		subnets = exec_io(
		[DOCKER, "network", "inspect"] + networks,
		deserialiser=JSONArray.from_string,
		).path(
		"$[].IPAM.Config[].Subnet", list[str],
		lambda subnets: {ipaddress.ip_network(net) for net in subnets},
		)
		for subnet in cls.DOCKER_SUBNET.subnets(8):
		if not any(net.overlaps(subnet) for net in subnets):
		return subnet
		raise LookupError(f"No free subnets found in subnet {cls.DOCKER_SUBNET}")

		def get_free_address(self) -> ipaddress.IPv4Address:
		"""
		Return a free address in the network