""" 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"): 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": {}} 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() # --------------------------------------------------------------------------- # create_ipvlan_network # --------------------------------------------------------------------------- class TestCreateIpvlanNetwork: def _make_client(self, existing=None, existing_driver="ipvlan"): 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": {}} 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 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")