neutron/tests/functional/agent/l3/test_namespace_manager.py

   1 # Copyright (c) 2015 Rackspace
   2 # All Rights Reserved.
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15
  16 import mock
  17 from oslo_utils import uuidutils
  18
  19 from neutron.agent.l3 import dvr_snat_ns
  20 from neutron.agent.l3 import namespace_manager
  21 from neutron.agent.l3 import namespaces
  22 from neutron.agent.linux import ip_lib
  23 from neutron.tests.functional import base
  24
  25 _uuid = uuidutils.generate_uuid
  26
  27
  28 class NamespaceManagerTestFramework(base.BaseSudoTestCase):
  29
  30     def setUp(self):
  31         super(NamespaceManagerTestFramework, self).setUp()
  32         self.agent_conf = mock.MagicMock()
  33         self.metadata_driver_mock = mock.Mock()
  34         self.namespace_manager = namespace_manager.NamespaceManager(
  35             self.agent_conf, driver=None,
  36             metadata_driver=self.metadata_driver_mock)
  37
  38     def _create_namespace(self, router_id, ns_class):
  39         namespace = ns_class(router_id, self.agent_conf, driver=None,
  40                              use_ipv6=False)
  41         namespace.create()
  42         self.addCleanup(self._delete_namespace, namespace)
  43         return namespace.name
  44
  45     def _delete_namespace(self, namespace):
  46         try:
  47             namespace.delete()
  48         except RuntimeError as e:
  49             # If the namespace didn't exist when delete was attempted, mission
  50             # accomplished. Otherwise, re-raise the exception
  51             if 'No such file or directory' not in str(e):
  52                 raise e
  53
  54     def _namespace_exists(self, namespace):
  55         ip = ip_lib.IPWrapper(namespace=namespace)
  56         return ip.netns.exists(namespace)
  57
  58
  59 class NamespaceManagerTestCase(NamespaceManagerTestFramework):
  60
  61     def test_namespace_manager(self):
  62         router_id = _uuid()
  63         router_id_to_delete = _uuid()
  64         to_keep = set()
  65         to_delete = set()
  66         to_retrieve = set()
  67         to_keep.add(self._create_namespace(router_id,
  68                                            namespaces.RouterNamespace))
  69         to_keep.add(self._create_namespace(router_id,
  70                                            dvr_snat_ns.SnatNamespace))
  71         to_delete.add(self._create_namespace(router_id_to_delete,
  72                                              dvr_snat_ns.SnatNamespace))
  73         to_retrieve = to_keep | to_delete
  74
  75         with mock.patch.object(namespace_manager.NamespaceManager, 'list_all',
  76                                return_value=to_retrieve):
  77             with self.namespace_manager as ns_manager:
  78                 for ns_name in to_keep:
  79                     id_to_keep = ns_manager.get_prefix_and_id(ns_name)[1]
  80                     ns_manager.keep_router(id_to_keep)
  81
  82         for ns_name in to_keep:
  83             self.assertTrue(self._namespace_exists(ns_name))
  84         for ns_name in to_delete:
  85             (self.metadata_driver_mock.destroy_monitored_metadata_proxy.
  86              assert_called_once_with(mock.ANY,
  87                                      router_id_to_delete,
  88                                      self.agent_conf))
  89             self.assertFalse(self._namespace_exists(ns_name))