anti/DECNET
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: testing → main (reconcile 2-week divergence)

Browse Source
This commit is contained in:
anti
2026-04-28 18:36:00 -04:00
parent 499836c9e4
commit 862e4dbb31
1235 changed files with 160255 additions and 7996 deletions
Download Patch File Download Diff File Expand all files Collapse all files

402
tests/core/test_network.py Normal file
View File

@@ -0,0 +1,402 @@
"""
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")