neutron/tests/unit/agent/l2/extensions/test_manager.py

   1 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   2 #    not use this file except in compliance with the License. You may obtain
   3 #    a copy of the License at
   4 #
   5 #         http://www.apache.org/licenses/LICENSE-2.0
   6 #
   7 #    Unless required by applicable law or agreed to in writing, software
   8 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
   9 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  10 #    License for the specific language governing permissions and limitations
  11 #    under the License.
  12
  13 import mock
  14 from oslo_config import cfg
  15
  16 from neutron.agent.l2.extensions import manager as ext_manager
  17 from neutron.tests import base
  18
  19
  20 class TestAgentExtensionsManager(base.BaseTestCase):
  21
  22     def setUp(self):
  23         super(TestAgentExtensionsManager, self).setUp()
  24         mock.patch('neutron.agent.l2.extensions.qos.QosAgentExtension',
  25                    autospec=True).start()
  26         conf = cfg.CONF
  27         ext_manager.register_opts(conf)
  28         cfg.CONF.set_override('extensions', ['qos'], 'agent')
  29         self.manager = ext_manager.AgentExtensionsManager(conf)
  30
  31     def _get_extension(self):
  32         return self.manager.extensions[0].obj
  33
  34     def test_initialize(self):
  35         connection = object()
  36         self.manager.initialize(connection, 'fake_driver_type')
  37         ext = self._get_extension()
  38         ext.initialize.assert_called_once_with(connection, 'fake_driver_type')
  39
  40     def test_handle_port(self):
  41         context = object()
  42         data = object()
  43         self.manager.handle_port(context, data)
  44         ext = self._get_extension()
  45         ext.handle_port.assert_called_once_with(context, data)
  46
  47     def test_delete_port(self):
  48         context = object()
  49         data = object()
  50         self.manager.delete_port(context, data)
  51         ext = self._get_extension()
  52         ext.delete_port.assert_called_once_with(context, data)