anti
91549e6936
Two compounding root causes produced the recurring 'Address already in use' error on redeploy: 1. _ensure_network only compared driver+name; if a prior deploy's IPAM pool drifted (different subnet/gateway/range), Docker kept handing out addresses from the old pool and raced the real LAN. Now also compares Subnet/Gateway/IPRange and rebuilds on drift. 2. A prior half-failed 'up' could leave containers still holding the IPs and ports the new run wants. Run 'compose down --remove-orphans' as a best-effort pre-up cleanup so IPAM starts from a clean state. Also surface docker compose stderr to the structured log on failure so the agent's journal captures Docker's actual message (which IP, which port) instead of just the exit code.
403 lines
17 KiB
Python
403 lines
17 KiB
Python
"""
|
||
Tests for decnet.network utility functions.
|
||
"""
|
||
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
|
||
from decnet.network import (
|
||
HOST_IPVLAN_IFACE,
|
||
HOST_MACVLAN_IFACE,
|
||
MACVLAN_NETWORK_NAME,
|
||
allocate_ips,
|
||
create_ipvlan_network,
|
||
create_macvlan_network,
|
||
detect_interface,
|
||
detect_subnet,
|
||
get_host_ip,
|
||
ips_to_range,
|
||
remove_macvlan_network,
|
||
setup_host_ipvlan,
|
||
setup_host_macvlan,
|
||
teardown_host_ipvlan,
|
||
teardown_host_macvlan,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# ips_to_range
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestIpsToRange:
|
||
def test_single_ip(self):
|
||
assert ips_to_range(["192.168.1.100"]) == "192.168.1.100/32"
|
||
|
||
def test_consecutive_small_range(self):
|
||
# .97–.101: max^min = 4, bit_length=3, prefix=29 → .96/29
|
||
result = ips_to_range([f"192.168.1.{i}" for i in range(97, 102)])
|
||
from ipaddress import IPv4Network, IPv4Address
|
||
net = IPv4Network(result)
|
||
for i in range(97, 102):
|
||
assert IPv4Address(f"192.168.1.{i}") in net
|
||
|
||
def test_range_crossing_cidr_boundary(self):
|
||
# .110–.119 crosses the /28 boundary (.96–.111 vs .112–.127)
|
||
# Subtraction gives /28 (wrong), XOR gives /27 (correct)
|
||
ips = [f"192.168.1.{i}" for i in range(110, 120)]
|
||
result = ips_to_range(ips)
|
||
from ipaddress import IPv4Network, IPv4Address
|
||
net = IPv4Network(result)
|
||
for i in range(110, 120):
|
||
assert IPv4Address(f"192.168.1.{i}") in net, (
|
||
f"192.168.1.{i} not in computed range {result}"
|
||
)
|
||
|
||
def test_all_ips_covered(self):
|
||
# Larger spread: .10–.200
|
||
ips = [f"10.0.0.{i}" for i in range(10, 201)]
|
||
result = ips_to_range(ips)
|
||
from ipaddress import IPv4Network, IPv4Address
|
||
net = IPv4Network(result)
|
||
for i in range(10, 201):
|
||
assert IPv4Address(f"10.0.0.{i}") in net
|
||
|
||
def test_two_ips_same_cidr(self):
|
||
# .100 and .101 share /31
|
||
result = ips_to_range(["192.168.1.100", "192.168.1.101"])
|
||
from ipaddress import IPv4Network, IPv4Address
|
||
net = IPv4Network(result)
|
||
assert IPv4Address("192.168.1.100") in net
|
||
assert IPv4Address("192.168.1.101") in net
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# create_macvlan_network
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestCreateMacvlanNetwork:
|
||
def _make_client(self, existing=None, existing_driver="macvlan",
|
||
ipam_subnet="192.168.1.0/24", ipam_gateway="192.168.1.1",
|
||
ipam_range="192.168.1.96/27"):
|
||
client = MagicMock()
|
||
nets = [MagicMock(name=n) for n in (existing or [])]
|
||
for net, n in zip(nets, (existing or [])):
|
||
net.name = n
|
||
net.attrs = {
|
||
"Driver": existing_driver,
|
||
"Containers": {},
|
||
"IPAM": {"Config": [{
|
||
"Subnet": ipam_subnet,
|
||
"Gateway": ipam_gateway,
|
||
"IPRange": ipam_range,
|
||
}]},
|
||
}
|
||
client.networks.list.return_value = nets
|
||
return client
|
||
|
||
def test_creates_network_when_absent(self):
|
||
client = self._make_client([])
|
||
create_macvlan_network(client, "eth0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
client.networks.create.assert_called_once()
|
||
kwargs = client.networks.create.call_args
|
||
assert kwargs[1]["driver"] == "macvlan"
|
||
assert kwargs[1]["name"] == MACVLAN_NETWORK_NAME
|
||
assert kwargs[1]["options"]["parent"] == "eth0"
|
||
|
||
def test_noop_when_network_exists(self):
|
||
client = self._make_client([MACVLAN_NETWORK_NAME])
|
||
create_macvlan_network(client, "eth0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
client.networks.create.assert_not_called()
|
||
|
||
def test_rebuilds_when_ipam_subnet_drifted(self):
|
||
"""Existing net matches driver+name, but IPAM pool is stale. Reusing it
|
||
hands out addresses from the old pool — surfaces as 'Address already in
|
||
use'. Must tear down + recreate."""
|
||
client = self._make_client([MACVLAN_NETWORK_NAME], ipam_subnet="10.0.0.0/24")
|
||
old_net = client.networks.list.return_value[0]
|
||
create_macvlan_network(client, "eth0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
old_net.remove.assert_called_once()
|
||
client.networks.create.assert_called_once()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# create_ipvlan_network
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestCreateIpvlanNetwork:
|
||
def _make_client(self, existing=None, existing_driver="ipvlan",
|
||
ipam_subnet="192.168.1.0/24", ipam_gateway="192.168.1.1",
|
||
ipam_range="192.168.1.96/27"):
|
||
client = MagicMock()
|
||
nets = [MagicMock(name=n) for n in (existing or [])]
|
||
for net, n in zip(nets, (existing or [])):
|
||
net.name = n
|
||
net.attrs = {
|
||
"Driver": existing_driver,
|
||
"Containers": {},
|
||
"IPAM": {"Config": [{
|
||
"Subnet": ipam_subnet,
|
||
"Gateway": ipam_gateway,
|
||
"IPRange": ipam_range,
|
||
}]},
|
||
}
|
||
client.networks.list.return_value = nets
|
||
return client
|
||
|
||
def test_creates_ipvlan_network(self):
|
||
client = self._make_client([])
|
||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
client.networks.create.assert_called_once()
|
||
kwargs = client.networks.create.call_args
|
||
assert kwargs[1]["driver"] == "ipvlan"
|
||
assert kwargs[1]["options"]["parent"] == "wlan0"
|
||
assert kwargs[1]["options"]["ipvlan_mode"] == "l2"
|
||
|
||
def test_noop_when_network_exists(self):
|
||
client = self._make_client([MACVLAN_NETWORK_NAME])
|
||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
client.networks.create.assert_not_called()
|
||
|
||
def test_replaces_macvlan_network_with_ipvlan(self):
|
||
"""If an old macvlan-driver net exists under the same name, remove+recreate.
|
||
Short-circuiting on name alone leaves Docker attaching containers to the
|
||
wrong driver — on the next port create the parent NIC goes EBUSY because
|
||
macvlan and ipvlan slaves can't share it."""
|
||
client = self._make_client([MACVLAN_NETWORK_NAME], existing_driver="macvlan")
|
||
old_net = client.networks.list.return_value[0]
|
||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
old_net.remove.assert_called_once()
|
||
client.networks.create.assert_called_once()
|
||
assert client.networks.create.call_args[1]["driver"] == "ipvlan"
|
||
|
||
def test_uses_same_network_name_as_macvlan(self):
|
||
"""Both drivers share the same logical network name so compose files are identical."""
|
||
client = self._make_client([])
|
||
create_ipvlan_network(client, "wlan0", "192.168.1.0/24", "192.168.1.1", "192.168.1.96/27")
|
||
assert client.networks.create.call_args[1]["name"] == MACVLAN_NETWORK_NAME
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# setup_host_macvlan / teardown_host_macvlan
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestSetupHostMacvlan:
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_creates_interface_when_absent(self, mock_run, _):
|
||
# Simulate interface not existing (returncode != 0)
|
||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||
calls = [str(c) for c in mock_run.call_args_list]
|
||
assert any("macvlan" in c for c in calls)
|
||
assert any("mode" in c and "bridge" in c for c in calls)
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_skips_create_when_interface_exists(self, mock_run, _):
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||
calls = [c[0][0] for c in mock_run.call_args_list]
|
||
# "ip link add <iface> link ..." should not be called when iface exists.
|
||
# (The opportunistic `ip link del decnet_ipvlan0` cleanup is allowed.)
|
||
add_cmds = [cmd for cmd in calls if cmd[:3] == ["ip", "link", "add"]]
|
||
assert not any(HOST_MACVLAN_IFACE in cmd for cmd in add_cmds)
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=1)
|
||
def test_requires_root(self, _):
|
||
with pytest.raises(PermissionError):
|
||
setup_host_macvlan("eth0", "192.168.1.5", "192.168.1.96/27")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# setup_host_ipvlan / teardown_host_ipvlan
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestSetupHostIpvlan:
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_creates_ipvlan_interface(self, mock_run, _):
|
||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||
calls = [str(c) for c in mock_run.call_args_list]
|
||
assert any("ipvlan" in c for c in calls)
|
||
assert any("mode" in c and "l2" in c for c in calls)
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_uses_ipvlan_iface_name(self, mock_run, _):
|
||
mock_run.side_effect = lambda cmd, **kw: MagicMock(returncode=1) if "show" in cmd else MagicMock(returncode=0)
|
||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||
calls = [c[0][0] for c in mock_run.call_args_list]
|
||
# Primary interface created is the ipvlan slave.
|
||
assert any("add" in cmd and HOST_IPVLAN_IFACE in cmd for cmd in calls)
|
||
# The only macvlan reference allowed is the opportunistic `del`
|
||
# that cleans up a stale helper from a prior macvlan deploy.
|
||
macvlan_refs = [cmd for cmd in calls if HOST_MACVLAN_IFACE in cmd]
|
||
assert all(cmd[:3] == ["ip", "link", "del"] for cmd in macvlan_refs)
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=1)
|
||
def test_requires_root(self, _):
|
||
with pytest.raises(PermissionError):
|
||
setup_host_ipvlan("wlan0", "192.168.1.5", "192.168.1.96/27")
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_teardown_uses_ipvlan_iface(self, mock_run, _):
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
teardown_host_ipvlan("192.168.1.96/27")
|
||
calls = [str(c) for c in mock_run.call_args_list]
|
||
assert any(HOST_IPVLAN_IFACE in c for c in calls)
|
||
assert not any(HOST_MACVLAN_IFACE in c for c in calls)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# allocate_ips (pure logic — no subprocess / Docker)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestAllocateIps:
|
||
def test_basic_allocation(self):
|
||
ips = allocate_ips("192.168.1.0/24", "192.168.1.1", "192.168.1.100", count=3)
|
||
assert len(ips) == 3
|
||
assert "192.168.1.1" not in ips # gateway skipped
|
||
assert "192.168.1.100" not in ips # host IP skipped
|
||
|
||
def test_skips_network_and_broadcast(self):
|
||
ips = allocate_ips("10.0.0.0/30", "10.0.0.1", "10.0.0.3", count=1)
|
||
# /30 hosts: .1 (gateway), .2. .3 is host_ip → only .2 available
|
||
assert ips == ["10.0.0.2"]
|
||
|
||
def test_respects_ip_start(self):
|
||
ips = allocate_ips("192.168.1.0/24", "192.168.1.1", "192.168.1.1",
|
||
count=2, ip_start="192.168.1.50")
|
||
assert all(ip >= "192.168.1.50" for ip in ips)
|
||
|
||
def test_raises_when_not_enough_ips(self):
|
||
# /30 only has 2 host addresses; reserving both leaves 0
|
||
with pytest.raises(RuntimeError, match="Not enough free IPs"):
|
||
allocate_ips("10.0.0.0/30", "10.0.0.1", "10.0.0.2", count=3)
|
||
|
||
def test_no_duplicates(self):
|
||
ips = allocate_ips("10.0.0.0/24", "10.0.0.1", "10.0.0.2", count=10)
|
||
assert len(ips) == len(set(ips))
|
||
|
||
def test_exact_count_returned(self):
|
||
ips = allocate_ips("172.16.0.0/24", "172.16.0.1", "172.16.0.254", count=5)
|
||
assert len(ips) == 5
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# detect_interface
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDetectInterface:
|
||
@patch("decnet.network._run")
|
||
def test_parses_dev_from_route(self, mock_run):
|
||
mock_run.return_value = MagicMock(
|
||
stdout="default via 192.168.1.1 dev eth0 proto dhcp\n"
|
||
)
|
||
assert detect_interface() == "eth0"
|
||
|
||
@patch("decnet.network._run")
|
||
def test_raises_when_no_dev_found(self, mock_run):
|
||
mock_run.return_value = MagicMock(stdout="")
|
||
with pytest.raises(RuntimeError, match="Could not auto-detect"):
|
||
detect_interface()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# detect_subnet
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestDetectSubnet:
|
||
def _make_run(self, addr_output, route_output):
|
||
def side_effect(cmd, **kwargs):
|
||
if "addr" in cmd:
|
||
return MagicMock(stdout=addr_output)
|
||
return MagicMock(stdout=route_output)
|
||
return side_effect
|
||
|
||
@patch("decnet.network._run")
|
||
def test_parses_subnet_and_gateway(self, mock_run):
|
||
mock_run.side_effect = self._make_run(
|
||
" inet 192.168.1.5/24 brd 192.168.1.255 scope global eth0\n",
|
||
"default via 192.168.1.1 dev eth0\n",
|
||
)
|
||
subnet, gw = detect_subnet("eth0")
|
||
assert subnet == "192.168.1.0/24"
|
||
assert gw == "192.168.1.1"
|
||
|
||
@patch("decnet.network._run")
|
||
def test_raises_when_no_inet(self, mock_run):
|
||
mock_run.side_effect = self._make_run("", "default via 192.168.1.1 dev eth0\n")
|
||
with pytest.raises(RuntimeError, match="Could not detect subnet"):
|
||
detect_subnet("eth0")
|
||
|
||
@patch("decnet.network._run")
|
||
def test_raises_when_no_gateway(self, mock_run):
|
||
mock_run.side_effect = self._make_run(
|
||
" inet 192.168.1.5/24 brd 192.168.1.255 scope global eth0\n", ""
|
||
)
|
||
with pytest.raises(RuntimeError, match="Could not detect gateway"):
|
||
detect_subnet("eth0")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# get_host_ip
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestGetHostIp:
|
||
@patch("decnet.network._run")
|
||
def test_returns_host_ip(self, mock_run):
|
||
mock_run.return_value = MagicMock(
|
||
stdout=" inet 10.0.0.5/24 brd 10.0.0.255 scope global eth0\n"
|
||
)
|
||
assert get_host_ip("eth0") == "10.0.0.5"
|
||
|
||
@patch("decnet.network._run")
|
||
def test_raises_when_no_inet(self, mock_run):
|
||
mock_run.return_value = MagicMock(stdout="link/ether aa:bb:cc:dd:ee:ff\n")
|
||
with pytest.raises(RuntimeError, match="Could not determine host IP"):
|
||
get_host_ip("eth0")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# remove_macvlan_network
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestRemoveMacvlanNetwork:
|
||
def test_removes_matching_network(self):
|
||
client = MagicMock()
|
||
net = MagicMock()
|
||
net.name = MACVLAN_NETWORK_NAME
|
||
client.networks.list.return_value = [net]
|
||
remove_macvlan_network(client)
|
||
net.remove.assert_called_once()
|
||
|
||
def test_noop_when_no_matching_network(self):
|
||
client = MagicMock()
|
||
other = MagicMock()
|
||
other.name = "some-other-network"
|
||
client.networks.list.return_value = [other]
|
||
remove_macvlan_network(client)
|
||
other.remove.assert_not_called()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# teardown_host_macvlan
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestTeardownHostMacvlan:
|
||
@patch("decnet.network.os.geteuid", return_value=0)
|
||
@patch("decnet.network._run")
|
||
def test_deletes_macvlan_iface(self, mock_run, _):
|
||
mock_run.return_value = MagicMock(returncode=0)
|
||
teardown_host_macvlan("192.168.1.96/27")
|
||
calls = [str(c) for c in mock_run.call_args_list]
|
||
assert any(HOST_MACVLAN_IFACE in c for c in calls)
|
||
|
||
@patch("decnet.network.os.geteuid", return_value=1)
|
||
def test_requires_root(self, _):
|
||
with pytest.raises(PermissionError):
|
||
teardown_host_macvlan("192.168.1.96/27")
|