x/vmware-nsx
Code Issues Proposed changes

Fix line endings from CRLF to LF.

Browse Source 
Fixes bug #1100234

Change-Id: Idb28558547b69ed4ad64334651405dc70fa9ddb3
This commit is contained in:
David Ripton 2013-01-16 09:14:19 -05:00
parent 20894c305c
commit d8b0f10ec7
5 changed files with 340 additions and 340 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
quantum/plugins/hyperv/common/__init__.py
View File

@ -1,16 +1,16 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions SRL # Copyright 2013 Cloudbase Solutions SRL
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.

32
quantum/tests/unit/hyperv/__init__.py
View File

@ -1,16 +1,16 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions SRL # Copyright 2013 Cloudbase Solutions SRL
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.

216
quantum/tests/unit/hyperv/test_hyperv_quantum_agent.py
View File

@ -1,113 +1,113 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions SRL # Copyright 2013 Cloudbase Solutions SRL
# Copyright 2013 Pedro Navarro Perez # Copyright 2013 Pedro Navarro Perez
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
""" """
Unit tests for Windows Hyper-V virtual switch quantum driver Unit tests for Windows Hyper-V virtual switch quantum driver
""" """
import sys import sys
import mock import mock
import unittest2 as unittest import unittest2 as unittest
from quantum.openstack.common import cfg from quantum.openstack.common import cfg
from quantum.plugins.hyperv.agent import hyperv_quantum_agent from quantum.plugins.hyperv.agent import hyperv_quantum_agent
class TestHyperVQuantumAgent(unittest.TestCase): class TestHyperVQuantumAgent(unittest.TestCase):
def setUp(self): def setUp(self):
self.addCleanup(cfg.CONF.reset) self.addCleanup(cfg.CONF.reset)
# Avoid rpc initialization for unit tests # Avoid rpc initialization for unit tests
cfg.CONF.set_override('rpc_backend', cfg.CONF.set_override('rpc_backend',
'quantum.openstack.common.rpc.impl_fake') 'quantum.openstack.common.rpc.impl_fake')
self.agent = hyperv_quantum_agent.HyperVQuantumAgent() self.agent = hyperv_quantum_agent.HyperVQuantumAgent()
self.agent.plugin_rpc = mock.Mock() self.agent.plugin_rpc = mock.Mock()
self.agent.context = mock.Mock() self.agent.context = mock.Mock()
self.agent.agent_id = mock.Mock() self.agent.agent_id = mock.Mock()
self.agent._utils = mock.Mock() self.agent._utils = mock.Mock()
def tearDown(self): def tearDown(self):
cfg.CONF.reset() cfg.CONF.reset()
def test_port_bound(self): def test_port_bound(self):
port = mock.Mock() port = mock.Mock()
net_uuid = 'my-net-uuid' net_uuid = 'my-net-uuid'
with mock.patch.object( with mock.patch.object(
self.agent._utils, 'connect_vnic_to_vswitch'): self.agent._utils, 'connect_vnic_to_vswitch'):
with mock.patch.object( with mock.patch.object(
self.agent._utils, 'set_vswitch_port_vlan_id'): self.agent._utils, 'set_vswitch_port_vlan_id'):
self.agent._port_bound(port, net_uuid, 'vlan', None, None) self.agent._port_bound(port, net_uuid, 'vlan', None, None)
def test_port_unbound(self): def test_port_unbound(self):
map = { map = {
'network_type': 'vlan', 'network_type': 'vlan',
'vswitch_name': 'fake-vswitch', 'vswitch_name': 'fake-vswitch',
'ports': [], 'ports': [],
'vlan_id': 1} 'vlan_id': 1}
net_uuid = 'my-net-uuid' net_uuid = 'my-net-uuid'
network_vswitch_map = (net_uuid, map) network_vswitch_map = (net_uuid, map)
with mock.patch.object(self.agent, with mock.patch.object(self.agent,
'_get_network_vswitch_map_by_port_id', '_get_network_vswitch_map_by_port_id',
return_value=network_vswitch_map): return_value=network_vswitch_map):
with mock.patch.object( with mock.patch.object(
self.agent._utils, self.agent._utils,
'disconnect_switch_port'): 'disconnect_switch_port'):
self.agent._port_unbound(net_uuid) self.agent._port_unbound(net_uuid)
def test_treat_devices_added_returns_true_for_missing_device(self): def test_treat_devices_added_returns_true_for_missing_device(self):
attrs = {'get_device_details.side_effect': Exception()} attrs = {'get_device_details.side_effect': Exception()}
self.agent.plugin_rpc.configure_mock(**attrs) self.agent.plugin_rpc.configure_mock(**attrs)
self.assertTrue(self.agent._treat_devices_added([{}])) self.assertTrue(self.agent._treat_devices_added([{}]))
def mock_treat_devices_added(self, details, func_name): def mock_treat_devices_added(self, details, func_name):
""" """
:param details: the details to return for the device :param details: the details to return for the device
:param func_name: the function that should be called :param func_name: the function that should be called
:returns: whether the named function was called :returns: whether the named function was called
""" """
attrs = {'get_device_details.return_value': details} attrs = {'get_device_details.return_value': details}
self.agent.plugin_rpc.configure_mock(**attrs) self.agent.plugin_rpc.configure_mock(**attrs)
with mock.patch.object(self.agent, func_name) as func: with mock.patch.object(self.agent, func_name) as func:
self.assertFalse(self.agent._treat_devices_added([{}])) self.assertFalse(self.agent._treat_devices_added([{}]))
return func.called return func.called
def test_treat_devices_added_updates_known_port(self): def test_treat_devices_added_updates_known_port(self):
details = mock.MagicMock() details = mock.MagicMock()
details.__contains__.side_effect = lambda x: True details.__contains__.side_effect = lambda x: True
self.assertTrue(self.mock_treat_devices_added(details, self.assertTrue(self.mock_treat_devices_added(details,
'_treat_vif_port')) '_treat_vif_port'))
def test_treat_devices_removed_returns_true_for_missing_device(self): def test_treat_devices_removed_returns_true_for_missing_device(self):
attrs = {'update_device_down.side_effect': Exception()} attrs = {'update_device_down.side_effect': Exception()}
self.agent.plugin_rpc.configure_mock(**attrs) self.agent.plugin_rpc.configure_mock(**attrs)
self.assertTrue(self.agent._treat_devices_removed([{}])) self.assertTrue(self.agent._treat_devices_removed([{}]))
def mock_treat_devices_removed(self, port_exists): def mock_treat_devices_removed(self, port_exists):
details = dict(exists=port_exists) details = dict(exists=port_exists)
attrs = {'update_device_down.return_value': details} attrs = {'update_device_down.return_value': details}
self.agent.plugin_rpc.configure_mock(**attrs) self.agent.plugin_rpc.configure_mock(**attrs)
with mock.patch.object(self.agent, '_port_unbound') as func: with mock.patch.object(self.agent, '_port_unbound') as func:
self.assertFalse(self.agent._treat_devices_removed([{}])) self.assertFalse(self.agent._treat_devices_removed([{}]))
self.assertEqual(func.called, not port_exists) self.assertEqual(func.called, not port_exists)
def test_treat_devices_removed_unbinds_port(self): def test_treat_devices_removed_unbinds_port(self):
self.mock_treat_devices_removed(False) self.mock_treat_devices_removed(False)
def test_treat_devices_removed_ignores_missing_port(self): def test_treat_devices_removed_ignores_missing_port(self):
self.mock_treat_devices_removed(False) self.mock_treat_devices_removed(False)

164
quantum/tests/unit/hyperv/test_hyperv_quantum_plugin.py
View File

@ -1,88 +1,88 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions SRL # Copyright 2013 Cloudbase Solutions SRL
# Copyright 2013 Pedro Navarro Perez # Copyright 2013 Pedro Navarro Perez
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import contextlib import contextlib
from quantum import context from quantum import context
from quantum.extensions import portbindings from quantum.extensions import portbindings
from quantum.manager import QuantumManager from quantum.manager import QuantumManager
from quantum.openstack.common import cfg from quantum.openstack.common import cfg
from quantum.tests.unit import test_db_plugin as test_plugin from quantum.tests.unit import test_db_plugin as test_plugin
class HyperVQuantumPluginTestCase(test_plugin.QuantumDbPluginV2TestCase): class HyperVQuantumPluginTestCase(test_plugin.QuantumDbPluginV2TestCase):
_plugin_name = ('quantum.plugins.hyperv.' _plugin_name = ('quantum.plugins.hyperv.'
'hyperv_quantum_plugin.HyperVQuantumPlugin') 'hyperv_quantum_plugin.HyperVQuantumPlugin')
def setUp(self): def setUp(self):
super(HyperVQuantumPluginTestCase, self).setUp(self._plugin_name) super(HyperVQuantumPluginTestCase, self).setUp(self._plugin_name)
class TestHyperVVirtualSwitchBasicGet( class TestHyperVVirtualSwitchBasicGet(
test_plugin.TestBasicGet, HyperVQuantumPluginTestCase): test_plugin.TestBasicGet, HyperVQuantumPluginTestCase):
pass pass
class TestHyperVVirtualSwitchV2HTTPResponse( class TestHyperVVirtualSwitchV2HTTPResponse(
test_plugin.TestV2HTTPResponse, HyperVQuantumPluginTestCase): test_plugin.TestV2HTTPResponse, HyperVQuantumPluginTestCase):
pass pass
class TestHyperVVirtualSwitchPortsV2( class TestHyperVVirtualSwitchPortsV2(
test_plugin.TestPortsV2, HyperVQuantumPluginTestCase): test_plugin.TestPortsV2, HyperVQuantumPluginTestCase):
def test_port_vif_details(self): def test_port_vif_details(self):
plugin = QuantumManager.get_plugin() plugin = QuantumManager.get_plugin()
with self.port(name='name') as port: with self.port(name='name') as port:
port_id = port['port']['id'] port_id = port['port']['id']
self.assertEqual(port['port']['binding:vif_type'], self.assertEqual(port['port']['binding:vif_type'],
portbindings.VIF_TYPE_HYPERV) portbindings.VIF_TYPE_HYPERV)
# By default user is admin - now test non admin user # By default user is admin - now test non admin user
ctx = context.Context(user_id=None, ctx = context.Context(user_id=None,
tenant_id=self._tenant_id, tenant_id=self._tenant_id,
is_admin=False, is_admin=False,
read_deleted="no") read_deleted="no")
non_admin_port = plugin.get_port(ctx, port_id) non_admin_port = plugin.get_port(ctx, port_id)
self.assertTrue('status' in non_admin_port) self.assertTrue('status' in non_admin_port)
self.assertFalse('binding:vif_type' in non_admin_port) self.assertFalse('binding:vif_type' in non_admin_port)
def test_ports_vif_details(self): def test_ports_vif_details(self):
cfg.CONF.set_default('allow_overlapping_ips', True) cfg.CONF.set_default('allow_overlapping_ips', True)
plugin = QuantumManager.get_plugin() plugin = QuantumManager.get_plugin()
with contextlib.nested(self.port(), self.port()) as (port1, port2): with contextlib.nested(self.port(), self.port()) as (port1, port2):
ctx = context.get_admin_context() ctx = context.get_admin_context()
ports = plugin.get_ports(ctx) ports = plugin.get_ports(ctx)
self.assertEqual(len(ports), 2) self.assertEqual(len(ports), 2)
for port in ports: for port in ports:
self.assertEqual(port['binding:vif_type'], self.assertEqual(port['binding:vif_type'],
portbindings.VIF_TYPE_HYPERV) portbindings.VIF_TYPE_HYPERV)
# By default user is admin - now test non admin user # By default user is admin - now test non admin user
ctx = context.Context(user_id=None, ctx = context.Context(user_id=None,
tenant_id=self._tenant_id, tenant_id=self._tenant_id,
is_admin=False, is_admin=False,
read_deleted="no") read_deleted="no")
ports = plugin.get_ports(ctx) ports = plugin.get_ports(ctx)
self.assertEqual(len(ports), 2) self.assertEqual(len(ports), 2)
for non_admin_port in ports: for non_admin_port in ports:
self.assertTrue('status' in non_admin_port) self.assertTrue('status' in non_admin_port)
self.assertFalse('binding:vif_type' in non_admin_port) self.assertFalse('binding:vif_type' in non_admin_port)
class TestHyperVVirtualSwitchNetworksV2( class TestHyperVVirtualSwitchNetworksV2(
test_plugin.TestNetworksV2, HyperVQuantumPluginTestCase): test_plugin.TestNetworksV2, HyperVQuantumPluginTestCase):
pass pass

236
quantum/tests/unit/hyperv/test_hyperv_rpcapi.py
View File

@ -1,126 +1,126 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4 # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 Cloudbase Solutions SRL # Copyright 2013 Cloudbase Solutions SRL
# Copyright 2013 Pedro Navarro Perez # Copyright 2013 Pedro Navarro Perez
# All Rights Reserved. # All Rights Reserved.
# #
# Licensed under the Apache License, Version 2.0 (the "License"); you may # 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 # not use this file except in compliance with the License. You may obtain
# a copy of the License at # a copy of the License at
# #
# http://www.apache.org/licenses/LICENSE-2.0 # http://www.apache.org/licenses/LICENSE-2.0
# #
# Unless required by applicable law or agreed to in writing, software # Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
""" """
Unit Tests for hyperv quantum rpc Unit Tests for hyperv quantum rpc
""" """
import mock import mock
import unittest2 import unittest2
from quantum.agent import rpc as agent_rpc from quantum.agent import rpc as agent_rpc
from quantum.common import topics from quantum.common import topics
from quantum.openstack.common import context from quantum.openstack.common import context
from quantum.openstack.common import rpc from quantum.openstack.common import rpc
from quantum.plugins.hyperv import agent_notifier_api as ana from quantum.plugins.hyperv import agent_notifier_api as ana
from quantum.plugins.hyperv.common import constants from quantum.plugins.hyperv.common import constants
class rpcHyperVApiTestCase(unittest2.TestCase): class rpcHyperVApiTestCase(unittest2.TestCase):
def _test_hyperv_quantum_api( def _test_hyperv_quantum_api(
self, rpcapi, topic, method, rpc_method, **kwargs): self, rpcapi, topic, method, rpc_method, **kwargs):
ctxt = context.RequestContext('fake_user', 'fake_project') ctxt = context.RequestContext('fake_user', 'fake_project')
expected_retval = 'foo' if method == 'call' else None expected_retval = 'foo' if method == 'call' else None
expected_msg = rpcapi.make_msg(method, **kwargs) expected_msg = rpcapi.make_msg(method, **kwargs)
expected_msg['version'] = rpcapi.BASE_RPC_API_VERSION expected_msg['version'] = rpcapi.BASE_RPC_API_VERSION
if rpc_method == 'cast' and method == 'run_instance': if rpc_method == 'cast' and method == 'run_instance':
kwargs['call'] = False kwargs['call'] = False
rpc_method_mock = mock.Mock() rpc_method_mock = mock.Mock()
rpc_method_mock.return_value = expected_retval rpc_method_mock.return_value = expected_retval
setattr(rpc, rpc_method, rpc_method_mock) setattr(rpc, rpc_method, rpc_method_mock)
retval = getattr(rpcapi, method)(ctxt, **kwargs) retval = getattr(rpcapi, method)(ctxt, **kwargs)
self.assertEqual(retval, expected_retval) self.assertEqual(retval, expected_retval)
expected_args = [ctxt, topic, expected_msg] expected_args = [ctxt, topic, expected_msg]
for arg, expected_arg in zip(rpc_method_mock.call_args[0], for arg, expected_arg in zip(rpc_method_mock.call_args[0],
expected_args): expected_args):
self.assertEqual(arg, expected_arg) self.assertEqual(arg, expected_arg)
def test_delete_network(self): def test_delete_network(self):
rpcapi = ana.AgentNotifierApi(topics.AGENT) rpcapi = ana.AgentNotifierApi(topics.AGENT)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, rpcapi,
topics.get_topic_name( topics.get_topic_name(
topics.AGENT, topics.AGENT,
topics.NETWORK, topics.NETWORK,
topics.DELETE), topics.DELETE),
'network_delete', rpc_method='fanout_cast', 'network_delete', rpc_method='fanout_cast',
network_id='fake_request_spec') network_id='fake_request_spec')
def test_port_update(self): def test_port_update(self):
rpcapi = ana.AgentNotifierApi(topics.AGENT) rpcapi = ana.AgentNotifierApi(topics.AGENT)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, rpcapi,
topics.get_topic_name( topics.get_topic_name(
topics.AGENT, topics.AGENT,
topics.PORT, topics.PORT,
topics.UPDATE), topics.UPDATE),
'port_update', rpc_method='fanout_cast', 'port_update', rpc_method='fanout_cast',
port='fake_port', port='fake_port',
network_type='fake_network_type', network_type='fake_network_type',
segmentation_id='fake_segmentation_id', segmentation_id='fake_segmentation_id',
physical_network='fake_physical_network') physical_network='fake_physical_network')
def test_port_delete(self): def test_port_delete(self):
rpcapi = ana.AgentNotifierApi(topics.AGENT) rpcapi = ana.AgentNotifierApi(topics.AGENT)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, rpcapi,
topics.get_topic_name( topics.get_topic_name(
topics.AGENT, topics.AGENT,
topics.PORT, topics.PORT,
topics.DELETE), topics.DELETE),
'port_delete', rpc_method='fanout_cast', 'port_delete', rpc_method='fanout_cast',
port_id='port_id') port_id='port_id')
def test_tunnel_update(self): def test_tunnel_update(self):
rpcapi = ana.AgentNotifierApi(topics.AGENT) rpcapi = ana.AgentNotifierApi(topics.AGENT)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, rpcapi,
topics.get_topic_name( topics.get_topic_name(
topics.AGENT, topics.AGENT,
constants.TUNNEL, constants.TUNNEL,
topics.UPDATE), topics.UPDATE),
'tunnel_update', rpc_method='fanout_cast', 'tunnel_update', rpc_method='fanout_cast',
tunnel_ip='fake_ip', tunnel_id='fake_id') tunnel_ip='fake_ip', tunnel_id='fake_id')
def test_device_details(self): def test_device_details(self):
rpcapi = agent_rpc.PluginApi(topics.PLUGIN) rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, topics.PLUGIN, rpcapi, topics.PLUGIN,
'get_device_details', rpc_method='call', 'get_device_details', rpc_method='call',
device='fake_device', device='fake_device',
agent_id='fake_agent_id') agent_id='fake_agent_id')
def test_update_device_down(self): def test_update_device_down(self):
rpcapi = agent_rpc.PluginApi(topics.PLUGIN) rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, topics.PLUGIN, rpcapi, topics.PLUGIN,
'update_device_down', rpc_method='call', 'update_device_down', rpc_method='call',
device='fake_device', device='fake_device',
agent_id='fake_agent_id') agent_id='fake_agent_id')
def test_tunnel_sync(self): def test_tunnel_sync(self):
rpcapi = agent_rpc.PluginApi(topics.PLUGIN) rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
self._test_hyperv_quantum_api( self._test_hyperv_quantum_api(
rpcapi, topics.PLUGIN, rpcapi, topics.PLUGIN,
'tunnel_sync', rpc_method='call', 'tunnel_sync', rpc_method='call',
tunnel_ip='fake_tunnel_ip') tunnel_ip='fake_tunnel_ip')