Add UT for agent_extensions_manager

Partially-Implements: blueprint quantum-qos-api
Change-Id: I86cf669dabbdad9680b6739d59e0f81a74c8629f
This commit is contained in:
Jakub Libosvar 2015-07-14 14:48:28 +00:00
parent 8aa404d15c
commit c240d381dc
2 changed files with 57 additions and 1 deletions

View File

@ -21,7 +21,6 @@ from neutron.i18n import _LE, _LI
LOG = log.getLogger(__name__)
# TODO(QoS) add unit tests to Agent extensions mgr
class AgentExtensionsManager(stevedore.named.NamedExtensionManager):
"""Manage agent extensions."""

View File

@ -0,0 +1,57 @@
# 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 mock
from neutron.agent.l2 import agent_extensions_manager
from neutron.tests import base
class TestAgentExtensionsManager(base.BaseTestCase):
def setUp(self):
super(TestAgentExtensionsManager, self).setUp()
mock.patch('neutron.agent.l2.extensions.qos_agent.QosAgentExtension',
autospec=True).start()
self.manager = agent_extensions_manager.AgentExtensionsManager()
def _get_extension(self):
return self.manager.extensions[0].obj
def test__call_on_agent_extension_missing_attribute_doesnt_crash(self):
self.manager._call_on_agent_extensions('foo', 'bar', 'baz')
def test_initialize(self):
self.manager.initialize()
ext = self._get_extension()
self.assertTrue(ext.initialize.called)
def test_handle_network(self):
context = object()
data = object()
self.manager.handle_network(context, data)
ext = self._get_extension()
ext.handle_network.assert_called_once_with(context, data)
def test_handle_subnet(self):
context = object()
data = object()
self.manager.handle_subnet(context, data)
ext = self._get_extension()
ext.handle_subnet.assert_called_once_with(context, data)
def test_handle_port(self):
context = object()
data = object()
self.manager.handle_port(context, data)
ext = self._get_extension()
ext.handle_port.assert_called_once_with(context, data)