Merge "Fix get_link_devices() with index argument"

neutron/privileged/agent/linux/ip_lib.py

@@ -619,9 +619,10 @@ def get_link_devices(namespace, **kwargs):
 
     :return: (list) interfaces in a namespace
     """
+    index = kwargs.pop('index') if 'index' in kwargs else 'all'
     try:
         with get_iproute(namespace) as ip:
-            return priv_linux.make_serializable(ip.get_links(**kwargs))
+            return priv_linux.make_serializable(ip.get_links(index, **kwargs))
     except OSError as e:
         if e.errno == errno.ENOENT:
             raise NetworkNamespaceNotFound(netns_name=namespace)

neutron/tests/functional/privileged/agent/linux/test_ip_lib.py

@@ -22,6 +22,7 @@
 from oslo_utils import uuidutils
 from pyroute2.ipdb import routes as ipdb_routes
 from pyroute2.iproute import linux as iproute_linux
+from pyroute2.netlink import exceptions as netlink_exc
 from pyroute2.netlink import rtnl
 import testtools
 
@@ -62,10 +63,10 @@ def test_get_device_names(self):
             self.assertNotIn(name, interfaces)
 
 
-class GetDevicesInfoTestCase(functional_base.BaseSudoTestCase):
+class GetLinkDevicesTestCase(functional_base.BaseSudoTestCase):
 
     def setUp(self):
-        super(GetDevicesInfoTestCase, self).setUp()
+        super().setUp()
         self.namespace = 'ns_test-' + uuidutils.generate_uuid()
         priv_ip_lib.create_netns(self.namespace)
         self.addCleanup(self._remove_ns, self.namespace)
@@ -76,7 +77,7 @@ def setUp(self):
     def _remove_ns(self, namespace):
         priv_ip_lib.remove_netns(namespace)
 
-    def test_get_devices_info_lo(self):
+    def test_get_link_devices_lo(self):
         devices = priv_ip_lib.get_link_devices(self.namespace)
         self.assertGreater(len(devices), 0)
         for device in devices:
@@ -87,7 +88,7 @@ def test_get_devices_info_lo(self):
         else:
             self.fail('Device "lo" not found')
 
-    def test_get_devices_info_dummy(self):
+    def test_get_link_devices_dummy(self):
         interfaces_tested = []
         for interface in self.interfaces:
             priv_ip_lib.create_interface(interface, self.namespace, 'dummy')
@@ -106,7 +107,7 @@ def test_get_devices_info_dummy(self):
             interfaces_tested.append(name)
         self.assertEqual(sorted(interfaces_tested), sorted(self.interfaces))
 
-    def test_get_devices_info_vlan(self):
+    def test_get_link_devices_vlan(self):
         interfaces_tested = []
         vlan_interfaces = []
         vlan_id = 1000
@@ -150,7 +151,7 @@ def test_get_devices_info_vlan(self):
         self.assertEqual(sorted(interfaces_tested),
                          sorted(self.interfaces + vlan_interfaces))
 
-    def test_get_devices_info_vxlan(self):
+    def test_get_link_devices_vxlan(self):
         interfaces_tested = []
         vxlan_interfaces = []
         vxlan_id = 1000
@@ -206,7 +207,7 @@ def _retrieve_interface(self, interface_name, namespace):
         else:
             self.fail('Interface "%s" not found' % interface_name)
 
-    def test_get_devices_info_veth_different_namespaces(self):
+    def test_get_link_devices_veth_different_namespaces(self):
         namespace2 = 'ns_test-' + uuidutils.generate_uuid()
         priv_ip_lib.create_netns(namespace2)
         self.addCleanup(self._remove_ns, namespace2)
@@ -231,7 +232,7 @@ def test_get_devices_info_veth_different_namespaces(self):
         self.assertIn(linux_utils.get_attr(veth1_1, 'IFLA_LINK'),
                       [None, veth1_2['index']])
 
-    def test_get_devices_info_veth_same_namespaces(self):
+    def test_get_link_devices_veth_same_namespaces(self):
         ip_wrapper = ip_lib.IPWrapper(self.namespace)
         ip_wrapper.add_veth('veth1_1', 'veth1_2')
 
@@ -243,6 +244,21 @@ def test_get_devices_info_veth_same_namespaces(self):
         self.assertEqual(veth1_1['index'], veth1_2_link)
         self.assertEqual(veth1_2['index'], veth1_1_link)
 
+    def test_get_link_devices_using_index(self):
+        for interface in self.interfaces:
+            priv_ip_lib.create_interface(interface, self.namespace, 'dummy')
+        for expected_device in priv_ip_lib.get_link_devices(self.namespace):
+            device = priv_ip_lib.get_link_devices(
+                self.namespace, index=expected_device['index'])
+            self.assertEqual(1, len(device))
+            self.assertEqual(linux_utils.get_attr(expected_device,
+                                                  'IFLA_IFNAME'),
+                             linux_utils.get_attr(device[0], 'IFLA_IFNAME'))
+
+        self.assertRaises(netlink_exc.NetlinkError,
+                          priv_ip_lib.get_link_devices, self.namespace,
+                          index=10000)
+
 
 class ListIpRulesTestCase(functional_base.BaseSudoTestCase):