text: # All Rights Reserved.
  #
  #    Licensed under the Apache License, Version 2.0 (the "License"); you may
  #    not use this file except in compliance with the License. You may obtain
  #    a copy of the License at
  #
  #         http://www.apache.org/licenses/LICENSE-2.0
  #
  #    Unless required by applicable law or agreed to in writing, software
  #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  #    License for the specific language governing permissions and limitations
  #    under the License.
  
  import errno
  
  import mock
  import pyroute2
  from pyroute2 import netlink
  from pyroute2.netlink.rtnl import ifinfmsg
  
  from neutron.privileged.agent.linux import ip_lib as priv_lib
  from neutron.tests import base
  
  
  class IpLibTestCase(base.BaseTestCase):
  
      def _test_run_iproute_link(self, namespace=None):
          ip_obj = "NetNS" if namespace else "IPRoute"
          with mock.patch.object(pyroute2, ip_obj) as ip_mock_cls:
              ip_mock = ip_mock_cls()
              ip_mock.__enter__().link_lookup.return_value = [2]
              priv_lib._run_iproute_link("test_cmd", "eth0", namespace,
                                         test_param="test_value")
              ip_mock.assert_has_calls([
                  mock.call.__enter__().link_lookup(ifname="eth0"),
                  mock.call.__exit__(None, None, None),
                  mock.call.__enter__().link("test_cmd", index=2,
                                             test_param="test_value")])
  
      def test_run_iproute_link_no_namespace(self):
          self._test_run_iproute_link()
  
      def test_run_iproute_link_in_namespace(self):
          self._test_run_iproute_link(namespace="testns")
  
      def test_run_iproute_link_interface_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ret_values = [
                  [],    # No interface found.
                  None,  # Unexpected output but also handled.
              ]
              for ret_val in ret_values:
                  ip_mock.__enter__().link_lookup.return_value = ret_val
                  self.assertRaises(
                      priv_lib.NetworkInterfaceNotFound,
                      priv_lib._run_iproute_link,
                      "test_cmd", "eth0", None, test_param="test_value")
  
      @mock.patch.object(priv_lib, 'get_iproute')
      def test_get_link_id(self, mock_iproute):
          mock_ip = mock.Mock()
          mock_ip.link_lookup.return_value = ['interface_id']
          mock_iproute.return_value.__enter__.return_value = mock_ip
          self.assertEqual('interface_id',
                           priv_lib.get_link_id('device', 'namespace'))
  
      def test_run_iproute_link_interface_removed_during_call(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = [2]
              ip_mock.__enter__().link.side_effect = pyroute2.NetlinkError(
                  code=errno.ENODEV)
              self.assertRaises(
                  priv_lib.NetworkInterfaceNotFound,
                  priv_lib._run_iproute_link,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_link_op_not_supported(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = [2]
              ip_mock.__enter__().link.side_effect = pyroute2.NetlinkError(
                  code=errno.EOPNOTSUPP)
              self.assertRaises(
                  priv_lib.InterfaceOperationNotSupported,
                  priv_lib._run_iproute_link,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_link_namespace_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.ENOENT, "Test no netns exception")
              self.assertRaises(
                  priv_lib.NetworkNamespaceNotFound,
                  priv_lib._run_iproute_link,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_link_error(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.EINVAL, "Test invalid argument exception")
              try:
                  priv_lib._run_iproute_link(
                      "test_cmd", "eth0", None, test_param="test_value")
                  self.fail("OSError exception not raised")
              except OSError as e:
                  self.assertEqual(errno.EINVAL, e.errno)
  
      def _test_run_iproute_neigh(self, namespace=None):
          ip_obj = "NetNS" if namespace else "IPRoute"
          with mock.patch.object(pyroute2, ip_obj) as ip_mock_cls:
              ip_mock = ip_mock_cls()
              ip_mock.__enter__().link_lookup.return_value = [2]
              priv_lib._run_iproute_neigh("test_cmd", "eth0", namespace,
                                          test_param="test_value")
              ip_mock.assert_has_calls([
                  mock.call.__enter__().link_lookup(ifname="eth0"),
                  mock.call.__exit__(None, None, None),
                  mock.call.__enter__().neigh("test_cmd", ifindex=2,
                                              test_param="test_value")])
  
      def test_run_iproute_neigh_no_namespace(self):
          self._test_run_iproute_neigh()
  
      def test_run_iproute_neigh_in_namespace(self):
          self._test_run_iproute_neigh(namespace="testns")
  
      def test_run_iproute_neigh_interface_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = []
              self.assertRaises(
                  priv_lib.NetworkInterfaceNotFound,
                  priv_lib._run_iproute_neigh,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_neigh_interface_removed_during_call(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = [2]
              ip_mock.__enter__().neigh.side_effect = pyroute2.NetlinkError(
                  code=errno.ENODEV)
              self.assertRaises(
                  priv_lib.NetworkInterfaceNotFound,
                  priv_lib._run_iproute_neigh,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_neigh_namespace_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.ENOENT, "Test no netns exception")
              self.assertRaises(
                  priv_lib.NetworkNamespaceNotFound,
                  priv_lib._run_iproute_neigh,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_neigh_error(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.EINVAL, "Test invalid argument exception")
              try:
                  priv_lib._run_iproute_neigh(
                      "test_cmd", "eth0", None, test_param="test_value")
                  self.fail("OSError exception not raised")
              except OSError as e:
                  self.assertEqual(errno.EINVAL, e.errno)
  
      def _test_run_iproute_addr(self, namespace=None):
          ip_obj = "NetNS" if namespace else "IPRoute"
          with mock.patch.object(pyroute2, ip_obj) as ip_mock_cls:
              ip_mock = ip_mock_cls()
              ip_mock.__enter__().link_lookup.return_value = [2]
              priv_lib._run_iproute_addr("test_cmd", "eth0", namespace,
                                         test_param="test_value")
              ip_mock.assert_has_calls([
                  mock.call.__enter__().link_lookup(ifname="eth0"),
                  mock.call.__exit__(None, None, None),
                  mock.call.__enter__().addr("test_cmd", index=2,
                                             test_param="test_value")])
  
      def test_run_iproute_addr_no_namespace(self):
          self._test_run_iproute_addr()
  
      def test_run_iproute_addr_in_namespace(self):
          self._test_run_iproute_addr(namespace="testns")
  
      def test_run_iproute_addr_interface_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = []
              self.assertRaises(
                  priv_lib.NetworkInterfaceNotFound,
                  priv_lib._run_iproute_addr,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_addr_interface_removed_during_call(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              ip_mock = iproute_mock()
              ip_mock.__enter__().link_lookup.return_value = [2]
              ip_mock.__enter__().addr.side_effect = pyroute2.NetlinkError(
                  code=errno.ENODEV)
              self.assertRaises(
                  priv_lib.NetworkInterfaceNotFound,
                  priv_lib._run_iproute_addr,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_addr_namespace_not_exists(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.ENOENT, "Test no netns exception")
              self.assertRaises(
                  priv_lib.NetworkNamespaceNotFound,
                  priv_lib._run_iproute_addr,
                  "test_cmd", "eth0", None, test_param="test_value")
  
      def test_run_iproute_addr_error(self):
          with mock.patch.object(pyroute2, "IPRoute") as iproute_mock:
              iproute_mock.side_effect = OSError(
                  errno.EINVAL, "Test invalid argument exception")
              try:
                  priv_lib._run_iproute_addr(
                      "test_cmd", "eth0", None, test_param="test_value")
                  self.fail("OSError exception not raised")
              except OSError as e:
                  self.assertEqual(errno.EINVAL, e.errno)
  
  
  class MakeSerializableTestCase(base.BaseTestCase):
  
      NLA_DATA1 = ifinfmsg.ifinfbase.state(data=b'54321')
      NLA_DATA2 = ifinfmsg.ifinfbase.state(data=b'abcdef')
      INPUT_1 = {'key1': 'value1', b'key2': b'value2', 'key3': ('a', 2),
                 'key4': [1, 2, 'c'],
                 b'key5': netlink.nla_slot('nla_name1', NLA_DATA1),
                 'key6': netlink.nla_slot(b'nla_name2', NLA_DATA2)}
      OUTPUT_1 = {'key1': 'value1', 'key2': 'value2', 'key3': ('a', 2),
                  'key4': [1, 2, 'c'],
                  'key5': ['nla_name1', '54321'],
                  'key6': ['nla_name2', 'abcdef']}
  
      def test_make_serializable(self):
          self.assertEqual(self.OUTPUT_1,
                           priv_lib.make_serializable(self.INPUT_1))
  