openstack
/
neutron
Code Issues Proposed changes

Use pyroute2 to check vlan/vxlan in use 

Now ip_lib.get_devices_info function is implemented using pyroute2,
"vlan_in_use" and "vxlan_in_use" can make use of it.

Change-Id: I82a2c3ea76195b10880cf37bf2229341b995b0ae
Closes-Bug: #1815498
This commit is contained in:
Rodolfo Alonso Hernandez 2019-02-12 10:05:28 +00:00
parent 53d3967827
commit cd31eae33d
2 changed files with 20 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
neutron/agent/linux/ip_lib.py
View File

@ -793,18 +793,18 @@ class IpNetnsCommand(IpCommandBase):
def vlan_in_use(segmentation_id, namespace=None):
"""Return True if VLAN ID is in use by an interface, else False."""
ip_wrapper = IPWrapper(namespace=namespace)
interfaces = ip_wrapper.netns.execute(["ip", "-d", "link", "list"],
check_exit_code=True)
return '802.1Q id %s ' % segmentation_id in interfaces
interfaces = get_devices_info(namespace)
vlans = {interface.get('vlan_id') for interface in interfaces
if interface.get('vlan_id')}
return segmentation_id in vlans
def vxlan_in_use(segmentation_id, namespace=None):
"""Return True if VXLAN VNID is in use by an interface, else False."""
ip_wrapper = IPWrapper(namespace=namespace)
interfaces = ip_wrapper.netns.execute(["ip", "-d", "link", "list"],
check_exit_code=True)
return 'vxlan id %s ' % segmentation_id in interfaces
interfaces = get_devices_info(namespace)
vxlans = {interface.get('vxlan_id') for interface in interfaces
if interface.get('vxlan_id')}
return segmentation_id in vxlans
def device_exists(device_name, namespace=None):

12
neutron/tests/functional/agent/linux/test_ip_lib.py
View File

@ -202,6 +202,18 @@ class IpLibTestCase(IpLibTestFramework):
device.link.delete()
self.assertFalse(device.exists())
def test_vlan_exists(self):
attr = self.generate_device_details()
ip = ip_lib.IPWrapper(namespace=attr.namespace)
ip.netns.add(attr.namespace)
self.addCleanup(ip.netns.delete, attr.namespace)
priv_ip_lib.create_interface(attr.name, attr.namespace, 'dummy')
self.assertFalse(ip_lib.vlan_in_use(1999, namespace=attr.namespace))
device = ip.add_vlan('vlan1999', attr.name, 1999)
self.assertTrue(ip_lib.vlan_in_use(1999, namespace=attr.namespace))
device.link.delete()
self.assertFalse(ip_lib.vlan_in_use(1999, namespace=attr.namespace))
def test_vxlan_exists(self):
attr = self.generate_device_details()
ip = ip_lib.IPWrapper(namespace=attr.namespace)