neutron-fwaas/neutron/tests/unit/services/vpn/test_vpn_agent.py
Salvatore Orlando dabbb1bfc5 Ensure unit tests do not let looping calls roam freely
Bug 1220871

This patch does minimal changes in neutron.plugins.nicira.common.sync
providing unit tests with a reference to the looping call object, so
that they can control its lifecycle.
Also, it perform a bit of refactoring in test_l3_agent.py in the way
mocks are created and started.

Change-Id: I73a1eb8ecdb7c6d46ff12afba549dd27095b7202
2013-09-04 17:19:21 -07:00

196 lines
7.0 KiB
Python

# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2013, Nachi Ueno, NTT I3, Inc.
# 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 mock
from oslo.config import cfg
from neutron.agent.common import config as agent_config
from neutron.agent import l3_agent
from neutron.agent.linux import interface
from neutron.common import config as base_config
from neutron.openstack.common import uuidutils
from neutron.services.vpn import agent
from neutron.services.vpn import device_drivers
from neutron.tests import base
_uuid = uuidutils.generate_uuid
NOOP_DEVICE_CLASS = 'NoopDeviceDriver'
NOOP_DEVICE = ('neutron.tests.unit.services.'
'vpn.test_vpn_agent.%s' % NOOP_DEVICE_CLASS)
class NoopDeviceDriver(device_drivers.DeviceDriver):
def sync(self, context, processes):
pass
def create_router(self, process_id):
pass
def destroy_router(self, process_id):
pass
class TestVPNAgent(base.BaseTestCase):
def setUp(self):
super(TestVPNAgent, self).setUp()
self.addCleanup(mock.patch.stopall)
self.conf = cfg.CONF
self.conf.register_opts(base_config.core_opts)
self.conf.register_opts(l3_agent.L3NATAgent.OPTS)
self.conf.register_opts(interface.OPTS)
agent_config.register_agent_state_opts_helper(self.conf)
agent_config.register_root_helper(self.conf)
self.conf.set_override('interface_driver',
'neutron.agent.linux.interface.NullDriver')
self.conf.set_override(
'vpn_device_driver',
[NOOP_DEVICE],
'vpnagent')
for clazz in [
'neutron.agent.linux.ip_lib.device_exists',
'neutron.agent.linux.ip_lib.IPWrapper',
'neutron.agent.linux.interface.NullDriver',
'neutron.agent.linux.utils.execute'
]:
mock.patch(clazz).start()
l3pluginApi_cls = mock.patch(
'neutron.agent.l3_agent.L3PluginApi').start()
self.plugin_api = mock.Mock()
l3pluginApi_cls.return_value = self.plugin_api
looping_call_p = mock.patch(
'neutron.openstack.common.loopingcall.FixedIntervalLoopingCall')
looping_call_p.start()
self.fake_host = 'fake_host'
self.agent = agent.VPNAgent(self.fake_host)
def test_setup_drivers(self):
self.assertEqual(1, len(self.agent.devices))
device = self.agent.devices[0]
self.assertEqual(
NOOP_DEVICE_CLASS,
device.__class__.__name__
)
def test_get_namespace(self):
router_id = _uuid()
ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
self.conf.use_namespaces, None)
self.agent.router_info = {router_id: ri}
namespace = self.agent.get_namespace(router_id)
self.assertTrue(namespace.endswith(router_id))
self.assertFalse(self.agent.get_namespace('fake_id'))
def test_add_nat_rule(self):
router_id = _uuid()
ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
self.conf.use_namespaces, None)
iptables = mock.Mock()
ri.iptables_manager.ipv4['nat'] = iptables
self.agent.router_info = {router_id: ri}
self.agent.add_nat_rule(router_id, 'fake_chain', 'fake_rule', True)
iptables.add_rule.assert_called_once_with(
'fake_chain', 'fake_rule', top=True)
def test_add_nat_rule_with_no_router(self):
self.agent.router_info = {}
#Should do nothing
self.agent.add_nat_rule(
'fake_router_id',
'fake_chain',
'fake_rule',
True)
def test_remove_rule(self):
router_id = _uuid()
ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
self.conf.use_namespaces, None)
iptables = mock.Mock()
ri.iptables_manager.ipv4['nat'] = iptables
self.agent.router_info = {router_id: ri}
self.agent.remove_nat_rule(router_id, 'fake_chain', 'fake_rule')
iptables.remove_rule.assert_called_once_with(
'fake_chain', 'fake_rule')
def test_remove_rule_with_no_router(self):
self.agent.router_info = {}
#Should do nothing
self.agent.remove_nat_rule(
'fake_router_id',
'fake_chain',
'fake_rule')
def test_iptables_apply(self):
router_id = _uuid()
ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
self.conf.use_namespaces, None)
iptables = mock.Mock()
ri.iptables_manager = iptables
self.agent.router_info = {router_id: ri}
self.agent.iptables_apply(router_id)
iptables.apply.assert_called_once_with()
def test_iptables_apply_with_no_router(self):
#Should do nothing
self.agent.router_info = {}
self.agent.iptables_apply('fake_router_id')
def test_router_added(self):
mock.patch(
'neutron.agent.linux.iptables_manager.IptablesManager').start()
router_id = _uuid()
router = {'id': router_id}
device = mock.Mock()
self.agent.devices = [device]
self.agent._router_added(router_id, router)
device.create_router.assert_called_once_with(router_id)
def test_router_removed(self):
self.plugin_api.get_external_network_id.return_value = None
mock.patch(
'neutron.agent.linux.iptables_manager.IptablesManager').start()
router_id = _uuid()
ri = l3_agent.RouterInfo(router_id, self.conf.root_helper,
self.conf.use_namespaces, None)
ri.router = {
'id': _uuid(),
'admin_state_up': True,
'routes': [],
'external_gateway_info': {}}
device = mock.Mock()
self.agent.router_info = {router_id: ri}
self.agent.devices = [device]
self.agent._router_removed(router_id)
device.destroy_router.assert_called_once_with(router_id)
def test_process_routers(self):
self.plugin_api.get_external_network_id.return_value = None
routers = [
{'id': _uuid(),
'admin_state_up': True,
'routes': [],
'external_gateway_info': {}}]
device = mock.Mock()
self.agent.devices = [device]
self.agent._process_routers(routers, False)
device.sync.assert_called_once_with(mock.ANY, routers)