1
0
Fork 0
mirror of https://github.com/retspen/webvirtcloud synced 2026-10-18 00:50:07 +00:00

refactor(vrtManager): replace vendored IPy with stdlib ipaddress and add tests

- Migrate network_size and wvmNetwork IP methods to Python built-in ipaddress
- Remove obsolete vendored vrtManager/IPy.py (1650+ lines)
- Add comprehensive unit test suite in vrtManager/test_network_ip.py (15 tests)
- Clean up obsolete IPy exclusion in .travis.yml
This commit is contained in:
catborise 2026-09-24 22:52:12 +03:00
parent ba9bc2c3b9
commit 6d8a62a95e
4 changed files with 185 additions and 1683 deletions

View file

@ -7,7 +7,7 @@ env:
install: install:
- pip install -r dev/requirements.txt - pip install -r dev/requirements.txt
script: script:
- pep8 --exclude=IPy.py --ignore=E501 vrtManager accounts admin appsettings \ - pep8 --ignore=E501 vrtManager accounts admin appsettings \
computesconsole create datasource instances \ computesconsole create datasource instances \
interfaceslogs networks nwfilters storages \ interfaceslogs networks nwfilters storages \
virtsecrets virtsecrets

File diff suppressed because it is too large Load diff

View file

@ -7,22 +7,23 @@ from libvirt import (
VIR_NETWORK_UPDATE_COMMAND_MODIFY, libvirtError) VIR_NETWORK_UPDATE_COMMAND_MODIFY, libvirtError)
from lxml import etree from lxml import etree
from vrtManager import util from vrtManager import util
import ipaddress
from vrtManager.connection import wvmConnect from vrtManager.connection import wvmConnect
from vrtManager.IPy import IP
def network_size(subnet, dhcp=None): def network_size(subnet, dhcp=None):
""" """
Func return gateway, mask and dhcp pool. Func return gateway, mask and dhcp pool.
""" """
mask = IP(subnet).strNetmask() net = ipaddress.ip_network(subnet, strict=False)
addr = IP(subnet) gateway = str(net.network_address + 1)
gateway = addr[1].strCompressed() if net.version == 4:
if addr.version() == 4: mask = str(net.netmask)
dhcp_pool = [addr[2].strCompressed(), addr[addr.len() - 2].strCompressed()] dhcp_pool = [str(net.network_address + 2), str(net.broadcast_address - 1)]
if addr.version() == 6: else:
mask = mask.lstrip("/") if "/" in mask else mask mask = str(net.prefixlen)
dhcp_pool = [IP(addr[0].strCompressed() + hex(256)), IP(addr[0].strCompressed() + hex(512 - 1))] dhcp_pool = [str(net.network_address + 256), str(net.network_address + 511)]
return (gateway, mask, dhcp_pool) if dhcp else (gateway, mask, None) return (gateway, mask, dhcp_pool) if dhcp else (gateway, mask, None)
@ -168,20 +169,13 @@ class wvmNetwork(wvmConnect):
netmask_str = ip.get("netmask") netmask_str = ip.get("netmask")
prefix = ip.get("prefix") prefix = ip.get("prefix")
family = ip.get("family", "ipv4") family = ip.get("family", "ipv4")
if prefix: mask = prefix if prefix is not None else netmask_str
prefix = int(prefix) if address_str and mask:
base = 32 if family == "ipv4" else 128 ret = ipaddress.ip_network(f"{address_str}/{mask}", strict=False)
binstr = prefix * "1" + (base - prefix) * "0" elif address_str:
netmask_str = str(IP(int(binstr, base=2))) ret = ipaddress.ip_address(address_str)
if netmask_str:
netmask = IP(netmask_str)
gateway = IP(address_str)
network = IP(gateway.int() & netmask.int())
netmask_str = netmask_str if family == "ipv4" else str(prefix)
ret = IP(str(network) + "/" + netmask_str)
else: else:
ret = IP(str(address_str)) continue
ip_networks[family] = ret ip_networks[family] = ret
return ip_networks return ip_networks
@ -204,7 +198,7 @@ class wvmNetwork(wvmConnect):
dhcpstart = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@start") dhcpstart = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@start")
dhcpend = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@end") dhcpend = util.get_xml_path(xml, "/network/ip[@family='ipv6']/dhcp/range[1]/@end")
return None if not dhcpstart or not dhcpend else [IP(dhcpstart), IP(dhcpend)] return None if not dhcpstart or not dhcpend else [ipaddress.ip_address(dhcpstart), ipaddress.ip_address(dhcpend)]
def get_dhcp_range_start(self, family="ipv4"): def get_dhcp_range_start(self, family="ipv4"):
dhcp = self.get_dhcp_range(family) dhcp = self.get_dhcp_range(family)
@ -282,12 +276,12 @@ class wvmNetwork(wvmConnect):
def modify_fixed_address(self, name, address, mac_duid, family="ipv4"): def modify_fixed_address(self, name, address, mac_duid, family="ipv4"):
tree = etree.fromstring(self._XMLDesc(0)) tree = etree.fromstring(self._XMLDesc(0))
if family == "ipv4": if family == "ipv4":
new_xml = '<host mac="{}" {} ip="{}"/>'.format(mac_duid, 'name="' + name + '"' if name else "", IP(address)) new_xml = '<host mac="{}" {} ip="{}"/>'.format(mac_duid, 'name="' + name + '"' if name else "", ipaddress.ip_address(address))
hosts = tree.xpath("./ip[not(@family='ipv6')]/dhcp/host") hosts = tree.xpath("./ip[not(@family='ipv6')]/dhcp/host")
compare_var = "mac" compare_var = "mac"
parent_index = self.parent_count - 2 parent_index = self.parent_count - 2
if family == "ipv6": if family == "ipv6":
new_xml = '<host id="{}" {} ip="{}"/>'.format(mac_duid, 'name="' + name + '"' if name else "", IP(address)) new_xml = '<host id="{}" {} ip="{}"/>'.format(mac_duid, 'name="' + name + '"' if name else "", ipaddress.ip_address(address))
hosts = tree.xpath("./ip[@family='ipv6']/dhcp/host") hosts = tree.xpath("./ip[@family='ipv6']/dhcp/host")
compare_var = "id" compare_var = "id"
parent_index = self.parent_count - 1 parent_index = self.parent_count - 1

View file

@ -0,0 +1,165 @@
import sys
import unittest
from unittest.mock import MagicMock
# Mock django if not installed so tests can run in minimal/standalone environments
if "django" not in sys.modules:
django_mock = MagicMock()
sys.modules["django"] = django_mock
sys.modules["django.conf"] = django_mock
from vrtManager.network import network_size, wvmNetwork
class TestNetworkSize(unittest.TestCase):
"""Unit tests for the network_size function."""
def test_ipv4_slash_24_with_dhcp(self):
gateway, mask, dhcp_pool = network_size("192.168.100.0/24", dhcp=True)
self.assertEqual(gateway, "192.168.100.1")
self.assertEqual(mask, "255.255.255.0")
self.assertIsNotNone(dhcp_pool)
self.assertEqual(str(dhcp_pool[0]), "192.168.100.2")
self.assertEqual(str(dhcp_pool[1]), "192.168.100.254")
def test_ipv4_slash_24_without_dhcp(self):
gateway, mask, dhcp_pool = network_size("192.168.100.0/24", dhcp=False)
self.assertEqual(gateway, "192.168.100.1")
self.assertEqual(mask, "255.255.255.0")
self.assertIsNone(dhcp_pool)
def test_ipv4_slash_16_with_dhcp(self):
gateway, mask, dhcp_pool = network_size("172.16.0.0/16", dhcp=True)
self.assertEqual(gateway, "172.16.0.1")
self.assertEqual(mask, "255.255.0.0")
self.assertIsNotNone(dhcp_pool)
self.assertEqual(str(dhcp_pool[0]), "172.16.0.2")
self.assertEqual(str(dhcp_pool[1]), "172.16.255.254")
def test_ipv4_slash_28_with_dhcp(self):
gateway, mask, dhcp_pool = network_size("10.0.0.0/28", dhcp=True)
self.assertEqual(gateway, "10.0.0.1")
self.assertEqual(mask, "255.255.255.240")
self.assertIsNotNone(dhcp_pool)
self.assertEqual(str(dhcp_pool[0]), "10.0.0.2")
self.assertEqual(str(dhcp_pool[1]), "10.0.0.14")
def test_ipv4_slash_30_with_dhcp(self):
gateway, mask, dhcp_pool = network_size("10.0.0.0/30", dhcp=True)
self.assertEqual(gateway, "10.0.0.1")
self.assertEqual(mask, "255.255.255.252")
self.assertIsNotNone(dhcp_pool)
self.assertEqual(str(dhcp_pool[0]), "10.0.0.2")
self.assertEqual(str(dhcp_pool[1]), "10.0.0.2")
def test_ipv6_slash_64_with_dhcp(self):
gateway, mask, dhcp_pool = network_size("2001:db8:ca2:2::/64", dhcp=True)
self.assertEqual(gateway, "2001:db8:ca2:2::1")
self.assertEqual(mask, "64")
self.assertIsNotNone(dhcp_pool)
self.assertEqual(str(dhcp_pool[0]), "2001:db8:ca2:2::100")
self.assertEqual(str(dhcp_pool[1]), "2001:db8:ca2:2::1ff")
def test_ipv6_slash_64_without_dhcp(self):
gateway, mask, dhcp_pool = network_size("2001:db8:ca2:2::/64", dhcp=False)
self.assertEqual(gateway, "2001:db8:ca2:2::1")
self.assertEqual(mask, "64")
self.assertIsNone(dhcp_pool)
def test_invalid_subnet(self):
with self.assertRaises(ValueError):
network_size("invalid.subnet/99")
class DummyNetwork(wvmNetwork):
"""Subclass of wvmNetwork with mock XML for isolated IP method testing."""
def __init__(self, xml_content):
self._xml = xml_content
self.parent_count = 2
self.updates = []
def _XMLDesc(self, flags):
return self._xml
def update(self, command, section, xml, parent_index, flags):
self.updates.append((command, section, xml, parent_index, flags))
class TestWvmNetworkIPOperations(unittest.TestCase):
"""Unit tests for IP parsing, DHCP ranges, and host addresses in wvmNetwork."""
DUAL_STACK_XML = """<network>
<name>default</name>
<ip address="192.168.100.1" netmask="255.255.255.0">
<dhcp>
<range start="192.168.100.2" end="192.168.100.254"/>
<host mac="52:54:00:12:34:56" name="vm1" ip="192.168.100.50"/>
</dhcp>
</ip>
<ip family="ipv6" address="2001:db8:ca2:2::1" prefix="64">
<dhcp>
<range start="2001:db8:ca2:2::100" end="2001:db8:ca2:2::1ff"/>
<host id="0:1:0:1" name="vm2" ip="2001:db8:ca2:2::50"/>
</dhcp>
</ip>
</network>"""
NO_IP_XML = """<network>
<name>isolated</name>
<bridge name="virbr1"/>
</network>"""
def test_get_ip_networks_dual_stack(self):
net = DummyNetwork(self.DUAL_STACK_XML)
ip_networks = net.get_ip_networks()
self.assertIn("ipv4", ip_networks)
self.assertIn("ipv6", ip_networks)
self.assertEqual(str(ip_networks["ipv4"]), "192.168.100.0/24")
self.assertEqual(str(ip_networks["ipv6"]), "2001:db8:ca2:2::/64")
def test_get_ip_networks_no_ip(self):
net = DummyNetwork(self.NO_IP_XML)
ip_networks = net.get_ip_networks()
self.assertEqual(ip_networks, {})
def test_get_dhcp_range_ipv4(self):
net = DummyNetwork(self.DUAL_STACK_XML)
dhcp = net.get_dhcp_range("ipv4")
self.assertIsNotNone(dhcp)
self.assertEqual(len(dhcp), 2)
self.assertEqual(str(dhcp[0]), "192.168.100.2")
self.assertEqual(str(dhcp[1]), "192.168.100.254")
def test_get_dhcp_range_ipv6(self):
net = DummyNetwork(self.DUAL_STACK_XML)
dhcp = net.get_dhcp_range("ipv6")
self.assertIsNotNone(dhcp)
self.assertEqual(len(dhcp), 2)
self.assertEqual(str(dhcp[0]), "2001:db8:ca2:2::100")
self.assertEqual(str(dhcp[1]), "2001:db8:ca2:2::1ff")
def test_get_dhcp_range_empty(self):
net = DummyNetwork(self.NO_IP_XML)
self.assertIsNone(net.get_dhcp_range("ipv4"))
self.assertIsNone(net.get_dhcp_range("ipv6"))
def test_get_dhcp_range_start_and_end(self):
net = DummyNetwork(self.DUAL_STACK_XML)
self.assertEqual(str(net.get_dhcp_range_start("ipv4")), "192.168.100.2")
self.assertEqual(str(net.get_dhcp_range_end("ipv4")), "192.168.100.254")
self.assertEqual(str(net.get_dhcp_range_start("ipv6")), "2001:db8:ca2:2::100")
self.assertEqual(str(net.get_dhcp_range_end("ipv6")), "2001:db8:ca2:2::1ff")
def test_modify_fixed_address_validation(self):
net = DummyNetwork(self.DUAL_STACK_XML)
# Invalid IP address must raise ValueError
with self.assertRaises(ValueError):
net.modify_fixed_address("bad_host", "999.999.999.999", "52:54:00:11:22:33", family="ipv4")
with self.assertRaises(ValueError):
net.modify_fixed_address("bad_host", "not-an-ip", "0:1:0:2", family="ipv6")
if __name__ == "__main__":
unittest.main()