Track Docker IP address assignments in Network (df6b082d) · Commits · Dom Sekotill / Behave Utils

behave_utils/docker.py

+29 −8

Original line number	Diff line number	Diff line
		@@ -329,7 +329,7 @@ class Container(Item):
		opts = [f'--alias={a}' for a in aliases]

		if address is None:
		address = network.get_free_address()
		address = network.reserve_address()
		opts.append(
		f"--ip={address}" if isinstance(address, ipaddress.IPv4Address) else
		f"--ip6={address}",
		@@ -404,6 +404,7 @@ class Network(Item):
		def __init__(self, name: str\|None = None) -> None:
		self._name = name or f"br{token_hex(6)}"
		self._nid: str\|None = None
		self._assigned = set[ipaddress.IPv4Address]()

		def __str__(self) -> str:
		return self._name
		@@ -462,7 +463,17 @@ class Network(Item):
		if len(data) == 0:
		raise
		self._nid = data.path("$[0].Id", str)
		self._assigned.update(
		data.path(
		"$[0].IPAM.Config[*].Gateway",
		list[str], lambda ls: (IPv4Address(s) for s in ls),
		),
		)
		else:
		self._assigned.add(gateway)
		assert self._nid is not None
		assert len(self._assigned) > 0, \
		"Expected gateways address(es) to be added to assigned addresses set"

		def destroy(self) -> None:
		"""
		@@ -492,7 +503,7 @@ class Network(Item):
		return subnet
		raise LookupError(f"No free subnets found in subnet {cls.DOCKER_SUBNET}")

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

		@@ -503,13 +514,23 @@ class Network(Item):
		data = self.inspect()
		# Considering only the first listed subnet
		net = data.path("$.IPAM.Config[0].Subnet", str, ipaddress.IPv4Network)
		reserved = data.path(

		# Recycle some old code for an assertion about assigned addresses
		if __debug__:
		reserved: set[ipaddress.IPv4Address] = data.path(
		"$.Containers.*.IPv4Address", list[str],
		lambda addrs: {IPv4Address.with_suffix(a) for a in addrs},
		)
		reserved.add(data.path("$.IPAM.Config[0].Gateway", str, IPv4Address))
		missing = reserved - self._assigned
		assert len(missing) == 0, f"Missing addresses from assigned set: {missing}"

		# Optimise for CPython 3.x without early binding
		assigned = self._assigned

		for addr in net.hosts():
		if addr not in reserved:
		if addr not in assigned:
		assigned.add(addr)
		return addr
		raise LookupError(f"No free addresses found in subnet {net}")