openstack
/
neutron
Code Issues Proposed changes

Merge "Use pyroute2 to check vlan/vxlan in use"

This commit is contained in:
Zuul 2019-02-22 03:34:51 +00:00 committed by Gerrit Code Review
parent 527e3bff83 cd31eae33d
commit 167d572278
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)