-# vim: tabstop=4 shiftwidth=4 softtabstop=4\r
-\r
-# Copyright 2013 Cloudbase Solutions SRL\r
-# All Rights Reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
-# vim: tabstop=4 shiftwidth=4 softtabstop=4\r
-\r
-# Copyright 2013 Cloudbase Solutions SRL\r
-# All Rights Reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# 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.
-# vim: tabstop=4 shiftwidth=4 softtabstop=4\r
-\r
-# Copyright 2013 Cloudbase Solutions SRL\r
-# Copyright 2013 Pedro Navarro Perez\r
-# All Rights Reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
-\r
-"""\r
-Unit tests for Windows Hyper-V virtual switch quantum driver\r
-"""\r
-\r
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# Copyright 2013 Pedro Navarro Perez
+# 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.
+
+"""
+Unit tests for Windows Hyper-V virtual switch quantum driver
+"""
+
import sys
-\r
+
import mock
import unittest2 as unittest
-\r
+
from quantum.openstack.common import cfg
from quantum.plugins.hyperv.agent import hyperv_quantum_agent
-\r
-\r
-class TestHyperVQuantumAgent(unittest.TestCase):\r
-\r
- def setUp(self):\r
- self.addCleanup(cfg.CONF.reset)\r
- # Avoid rpc initialization for unit tests\r
- cfg.CONF.set_override('rpc_backend',\r
- 'quantum.openstack.common.rpc.impl_fake')\r
- self.agent = hyperv_quantum_agent.HyperVQuantumAgent()\r
- self.agent.plugin_rpc = mock.Mock()\r
- self.agent.context = mock.Mock()\r
- self.agent.agent_id = mock.Mock()\r
- self.agent._utils = mock.Mock()\r
-\r
- def tearDown(self):\r
- cfg.CONF.reset()\r
-\r
- def test_port_bound(self):\r
- port = mock.Mock()\r
- net_uuid = 'my-net-uuid'\r
- with mock.patch.object(\r
- self.agent._utils, 'connect_vnic_to_vswitch'):\r
- with mock.patch.object(\r
- self.agent._utils, 'set_vswitch_port_vlan_id'):\r
- self.agent._port_bound(port, net_uuid, 'vlan', None, None)\r
-\r
- def test_port_unbound(self):\r
- map = {\r
- 'network_type': 'vlan',\r
- 'vswitch_name': 'fake-vswitch',\r
- 'ports': [],\r
- 'vlan_id': 1}\r
- net_uuid = 'my-net-uuid'\r
- network_vswitch_map = (net_uuid, map)\r
- with mock.patch.object(self.agent,\r
- '_get_network_vswitch_map_by_port_id',\r
- return_value=network_vswitch_map):\r
- with mock.patch.object(\r
- self.agent._utils,\r
- 'disconnect_switch_port'):\r
- self.agent._port_unbound(net_uuid)\r
-\r
- def test_treat_devices_added_returns_true_for_missing_device(self):\r
- attrs = {'get_device_details.side_effect': Exception()}\r
- self.agent.plugin_rpc.configure_mock(**attrs)\r
- self.assertTrue(self.agent._treat_devices_added([{}]))\r
-\r
- def mock_treat_devices_added(self, details, func_name):\r
- """\r
- :param details: the details to return for the device\r
- :param func_name: the function that should be called\r
- :returns: whether the named function was called\r
- """\r
- attrs = {'get_device_details.return_value': details}\r
- self.agent.plugin_rpc.configure_mock(**attrs)\r
- with mock.patch.object(self.agent, func_name) as func:\r
- self.assertFalse(self.agent._treat_devices_added([{}]))\r
- return func.called\r
-\r
- def test_treat_devices_added_updates_known_port(self):\r
- details = mock.MagicMock()\r
- details.__contains__.side_effect = lambda x: True\r
- self.assertTrue(self.mock_treat_devices_added(details,\r
- '_treat_vif_port'))\r
-\r
- def test_treat_devices_removed_returns_true_for_missing_device(self):\r
- attrs = {'update_device_down.side_effect': Exception()}\r
- self.agent.plugin_rpc.configure_mock(**attrs)\r
- self.assertTrue(self.agent._treat_devices_removed([{}]))\r
-\r
- def mock_treat_devices_removed(self, port_exists):\r
- details = dict(exists=port_exists)\r
- attrs = {'update_device_down.return_value': details}\r
- self.agent.plugin_rpc.configure_mock(**attrs)\r
- with mock.patch.object(self.agent, '_port_unbound') as func:\r
- self.assertFalse(self.agent._treat_devices_removed([{}]))\r
- self.assertEqual(func.called, not port_exists)\r
-\r
- def test_treat_devices_removed_unbinds_port(self):\r
- self.mock_treat_devices_removed(False)\r
-\r
- def test_treat_devices_removed_ignores_missing_port(self):\r
- self.mock_treat_devices_removed(False)\r
+
+
+class TestHyperVQuantumAgent(unittest.TestCase):
+
+ def setUp(self):
+ self.addCleanup(cfg.CONF.reset)
+ # Avoid rpc initialization for unit tests
+ cfg.CONF.set_override('rpc_backend',
+ 'quantum.openstack.common.rpc.impl_fake')
+ self.agent = hyperv_quantum_agent.HyperVQuantumAgent()
+ self.agent.plugin_rpc = mock.Mock()
+ self.agent.context = mock.Mock()
+ self.agent.agent_id = mock.Mock()
+ self.agent._utils = mock.Mock()
+
+ def tearDown(self):
+ cfg.CONF.reset()
+
+ def test_port_bound(self):
+ port = mock.Mock()
+ net_uuid = 'my-net-uuid'
+ with mock.patch.object(
+ self.agent._utils, 'connect_vnic_to_vswitch'):
+ with mock.patch.object(
+ self.agent._utils, 'set_vswitch_port_vlan_id'):
+ self.agent._port_bound(port, net_uuid, 'vlan', None, None)
+
+ def test_port_unbound(self):
+ map = {
+ 'network_type': 'vlan',
+ 'vswitch_name': 'fake-vswitch',
+ 'ports': [],
+ 'vlan_id': 1}
+ net_uuid = 'my-net-uuid'
+ network_vswitch_map = (net_uuid, map)
+ with mock.patch.object(self.agent,
+ '_get_network_vswitch_map_by_port_id',
+ return_value=network_vswitch_map):
+ with mock.patch.object(
+ self.agent._utils,
+ 'disconnect_switch_port'):
+ self.agent._port_unbound(net_uuid)
+
+ def test_treat_devices_added_returns_true_for_missing_device(self):
+ attrs = {'get_device_details.side_effect': Exception()}
+ self.agent.plugin_rpc.configure_mock(**attrs)
+ self.assertTrue(self.agent._treat_devices_added([{}]))
+
+ def mock_treat_devices_added(self, details, func_name):
+ """
+ :param details: the details to return for the device
+ :param func_name: the function that should be called
+ :returns: whether the named function was called
+ """
+ attrs = {'get_device_details.return_value': details}
+ self.agent.plugin_rpc.configure_mock(**attrs)
+ with mock.patch.object(self.agent, func_name) as func:
+ self.assertFalse(self.agent._treat_devices_added([{}]))
+ return func.called
+
+ def test_treat_devices_added_updates_known_port(self):
+ details = mock.MagicMock()
+ details.__contains__.side_effect = lambda x: True
+ self.assertTrue(self.mock_treat_devices_added(details,
+ '_treat_vif_port'))
+
+ def test_treat_devices_removed_returns_true_for_missing_device(self):
+ attrs = {'update_device_down.side_effect': Exception()}
+ self.agent.plugin_rpc.configure_mock(**attrs)
+ self.assertTrue(self.agent._treat_devices_removed([{}]))
+
+ def mock_treat_devices_removed(self, port_exists):
+ details = dict(exists=port_exists)
+ attrs = {'update_device_down.return_value': details}
+ self.agent.plugin_rpc.configure_mock(**attrs)
+ with mock.patch.object(self.agent, '_port_unbound') as func:
+ self.assertFalse(self.agent._treat_devices_removed([{}]))
+ self.assertEqual(func.called, not port_exists)
+
+ def test_treat_devices_removed_unbinds_port(self):
+ self.mock_treat_devices_removed(False)
+
+ def test_treat_devices_removed_ignores_missing_port(self):
+ self.mock_treat_devices_removed(False)
-# vim: tabstop=4 shiftwidth=4 softtabstop=4\r
-\r
-# Copyright 2013 Cloudbase Solutions SRL\r
-# Copyright 2013 Pedro Navarro Perez\r
-# All Rights Reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
-\r
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# Copyright 2013 Pedro Navarro Perez
+# 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 contextlib
-\r
+
from quantum import context
from quantum.extensions import portbindings
from quantum.manager import QuantumManager
from quantum.openstack.common import cfg
from quantum.tests.unit import test_db_plugin as test_plugin
-\r
-\r
-class HyperVQuantumPluginTestCase(test_plugin.QuantumDbPluginV2TestCase):\r
-\r
- _plugin_name = ('quantum.plugins.hyperv.'\r
- 'hyperv_quantum_plugin.HyperVQuantumPlugin')\r
-\r
- def setUp(self):\r
- super(HyperVQuantumPluginTestCase, self).setUp(self._plugin_name)\r
-\r
-\r
-class TestHyperVVirtualSwitchBasicGet(\r
- test_plugin.TestBasicGet, HyperVQuantumPluginTestCase):\r
- pass\r
-\r
-\r
-class TestHyperVVirtualSwitchV2HTTPResponse(\r
- test_plugin.TestV2HTTPResponse, HyperVQuantumPluginTestCase):\r
- pass\r
-\r
-\r
-class TestHyperVVirtualSwitchPortsV2(\r
- test_plugin.TestPortsV2, HyperVQuantumPluginTestCase):\r
- def test_port_vif_details(self):\r
- plugin = QuantumManager.get_plugin()\r
- with self.port(name='name') as port:\r
- port_id = port['port']['id']\r
- self.assertEqual(port['port']['binding:vif_type'],\r
- portbindings.VIF_TYPE_HYPERV)\r
- # By default user is admin - now test non admin user\r
- ctx = context.Context(user_id=None,\r
- tenant_id=self._tenant_id,\r
- is_admin=False,\r
- read_deleted="no")\r
- non_admin_port = plugin.get_port(ctx, port_id)\r
- self.assertTrue('status' in non_admin_port)\r
- self.assertFalse('binding:vif_type' in non_admin_port)\r
-\r
- def test_ports_vif_details(self):\r
- cfg.CONF.set_default('allow_overlapping_ips', True)\r
- plugin = QuantumManager.get_plugin()\r
- with contextlib.nested(self.port(), self.port()) as (port1, port2):\r
- ctx = context.get_admin_context()\r
- ports = plugin.get_ports(ctx)\r
- self.assertEqual(len(ports), 2)\r
- for port in ports:\r
- self.assertEqual(port['binding:vif_type'],\r
- portbindings.VIF_TYPE_HYPERV)\r
- # By default user is admin - now test non admin user\r
- ctx = context.Context(user_id=None,\r
- tenant_id=self._tenant_id,\r
- is_admin=False,\r
- read_deleted="no")\r
- ports = plugin.get_ports(ctx)\r
- self.assertEqual(len(ports), 2)\r
- for non_admin_port in ports:\r
- self.assertTrue('status' in non_admin_port)\r
- self.assertFalse('binding:vif_type' in non_admin_port)\r
-\r
-\r
-class TestHyperVVirtualSwitchNetworksV2(\r
- test_plugin.TestNetworksV2, HyperVQuantumPluginTestCase):\r
- pass\r
+
+
+class HyperVQuantumPluginTestCase(test_plugin.QuantumDbPluginV2TestCase):
+
+ _plugin_name = ('quantum.plugins.hyperv.'
+ 'hyperv_quantum_plugin.HyperVQuantumPlugin')
+
+ def setUp(self):
+ super(HyperVQuantumPluginTestCase, self).setUp(self._plugin_name)
+
+
+class TestHyperVVirtualSwitchBasicGet(
+ test_plugin.TestBasicGet, HyperVQuantumPluginTestCase):
+ pass
+
+
+class TestHyperVVirtualSwitchV2HTTPResponse(
+ test_plugin.TestV2HTTPResponse, HyperVQuantumPluginTestCase):
+ pass
+
+
+class TestHyperVVirtualSwitchPortsV2(
+ test_plugin.TestPortsV2, HyperVQuantumPluginTestCase):
+ def test_port_vif_details(self):
+ plugin = QuantumManager.get_plugin()
+ with self.port(name='name') as port:
+ port_id = port['port']['id']
+ self.assertEqual(port['port']['binding:vif_type'],
+ portbindings.VIF_TYPE_HYPERV)
+ # By default user is admin - now test non admin user
+ ctx = context.Context(user_id=None,
+ tenant_id=self._tenant_id,
+ is_admin=False,
+ read_deleted="no")
+ non_admin_port = plugin.get_port(ctx, port_id)
+ self.assertTrue('status' in non_admin_port)
+ self.assertFalse('binding:vif_type' in non_admin_port)
+
+ def test_ports_vif_details(self):
+ cfg.CONF.set_default('allow_overlapping_ips', True)
+ plugin = QuantumManager.get_plugin()
+ with contextlib.nested(self.port(), self.port()) as (port1, port2):
+ ctx = context.get_admin_context()
+ ports = plugin.get_ports(ctx)
+ self.assertEqual(len(ports), 2)
+ for port in ports:
+ self.assertEqual(port['binding:vif_type'],
+ portbindings.VIF_TYPE_HYPERV)
+ # By default user is admin - now test non admin user
+ ctx = context.Context(user_id=None,
+ tenant_id=self._tenant_id,
+ is_admin=False,
+ read_deleted="no")
+ ports = plugin.get_ports(ctx)
+ self.assertEqual(len(ports), 2)
+ for non_admin_port in ports:
+ self.assertTrue('status' in non_admin_port)
+ self.assertFalse('binding:vif_type' in non_admin_port)
+
+
+class TestHyperVVirtualSwitchNetworksV2(
+ test_plugin.TestNetworksV2, HyperVQuantumPluginTestCase):
+ pass
-# vim: tabstop=4 shiftwidth=4 softtabstop=4\r
-\r
-# Copyright 2013 Cloudbase Solutions SRL\r
-# Copyright 2013 Pedro Navarro Perez\r
-# All Rights Reserved.\r
-#\r
-# Licensed under the Apache License, Version 2.0 (the "License"); you may\r
-# not use this file except in compliance with the License. You may obtain\r
-# a copy of the License at\r
-#\r
-# http://www.apache.org/licenses/LICENSE-2.0\r
-#\r
-# Unless required by applicable law or agreed to in writing, software\r
-# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT\r
-# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the\r
-# License for the specific language governing permissions and limitations\r
-# under the License.\r
-\r
-"""\r
-Unit Tests for hyperv quantum rpc\r
-"""\r
-\r
+# vim: tabstop=4 shiftwidth=4 softtabstop=4
+
+# Copyright 2013 Cloudbase Solutions SRL
+# Copyright 2013 Pedro Navarro Perez
+# 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.
+
+"""
+Unit Tests for hyperv quantum rpc
+"""
+
import mock
import unittest2
-\r
+
from quantum.agent import rpc as agent_rpc
from quantum.common import topics
from quantum.openstack.common import context
from quantum.openstack.common import rpc
from quantum.plugins.hyperv import agent_notifier_api as ana
from quantum.plugins.hyperv.common import constants
-\r
-\r
-class rpcHyperVApiTestCase(unittest2.TestCase):\r
-\r
- def _test_hyperv_quantum_api(\r
- self, rpcapi, topic, method, rpc_method, **kwargs):\r
- ctxt = context.RequestContext('fake_user', 'fake_project')\r
- expected_retval = 'foo' if method == 'call' else None\r
- expected_msg = rpcapi.make_msg(method, **kwargs)\r
- expected_msg['version'] = rpcapi.BASE_RPC_API_VERSION\r
- if rpc_method == 'cast' and method == 'run_instance':\r
- kwargs['call'] = False\r
-\r
- rpc_method_mock = mock.Mock()\r
- rpc_method_mock.return_value = expected_retval\r
- setattr(rpc, rpc_method, rpc_method_mock)\r
-\r
- retval = getattr(rpcapi, method)(ctxt, **kwargs)\r
-\r
- self.assertEqual(retval, expected_retval)\r
-\r
- expected_args = [ctxt, topic, expected_msg]\r
- for arg, expected_arg in zip(rpc_method_mock.call_args[0],\r
- expected_args):\r
- self.assertEqual(arg, expected_arg)\r
-\r
- def test_delete_network(self):\r
- rpcapi = ana.AgentNotifierApi(topics.AGENT)\r
- self._test_hyperv_quantum_api(\r
- rpcapi,\r
- topics.get_topic_name(\r
- topics.AGENT,\r
- topics.NETWORK,\r
- topics.DELETE),\r
- 'network_delete', rpc_method='fanout_cast',\r
- network_id='fake_request_spec')\r
-\r
- def test_port_update(self):\r
- rpcapi = ana.AgentNotifierApi(topics.AGENT)\r
- self._test_hyperv_quantum_api(\r
- rpcapi,\r
- topics.get_topic_name(\r
- topics.AGENT,\r
- topics.PORT,\r
- topics.UPDATE),\r
- 'port_update', rpc_method='fanout_cast',\r
- port='fake_port',\r
- network_type='fake_network_type',\r
- segmentation_id='fake_segmentation_id',\r
- physical_network='fake_physical_network')\r
-\r
- def test_port_delete(self):\r
- rpcapi = ana.AgentNotifierApi(topics.AGENT)\r
- self._test_hyperv_quantum_api(\r
- rpcapi,\r
- topics.get_topic_name(\r
- topics.AGENT,\r
- topics.PORT,\r
- topics.DELETE),\r
- 'port_delete', rpc_method='fanout_cast',\r
- port_id='port_id')\r
-\r
- def test_tunnel_update(self):\r
- rpcapi = ana.AgentNotifierApi(topics.AGENT)\r
- self._test_hyperv_quantum_api(\r
- rpcapi,\r
- topics.get_topic_name(\r
- topics.AGENT,\r
- constants.TUNNEL,\r
- topics.UPDATE),\r
- 'tunnel_update', rpc_method='fanout_cast',\r
- tunnel_ip='fake_ip', tunnel_id='fake_id')\r
-\r
- def test_device_details(self):\r
- rpcapi = agent_rpc.PluginApi(topics.PLUGIN)\r
- self._test_hyperv_quantum_api(\r
- rpcapi, topics.PLUGIN,\r
- 'get_device_details', rpc_method='call',\r
- device='fake_device',\r
- agent_id='fake_agent_id')\r
-\r
- def test_update_device_down(self):\r
- rpcapi = agent_rpc.PluginApi(topics.PLUGIN)\r
- self._test_hyperv_quantum_api(\r
- rpcapi, topics.PLUGIN,\r
- 'update_device_down', rpc_method='call',\r
- device='fake_device',\r
- agent_id='fake_agent_id')\r
-\r
- def test_tunnel_sync(self):\r
- rpcapi = agent_rpc.PluginApi(topics.PLUGIN)\r
- self._test_hyperv_quantum_api(\r
- rpcapi, topics.PLUGIN,\r
- 'tunnel_sync', rpc_method='call',\r
- tunnel_ip='fake_tunnel_ip')\r
+
+
+class rpcHyperVApiTestCase(unittest2.TestCase):
+
+ def _test_hyperv_quantum_api(
+ self, rpcapi, topic, method, rpc_method, **kwargs):
+ ctxt = context.RequestContext('fake_user', 'fake_project')
+ expected_retval = 'foo' if method == 'call' else None
+ expected_msg = rpcapi.make_msg(method, **kwargs)
+ expected_msg['version'] = rpcapi.BASE_RPC_API_VERSION
+ if rpc_method == 'cast' and method == 'run_instance':
+ kwargs['call'] = False
+
+ rpc_method_mock = mock.Mock()
+ rpc_method_mock.return_value = expected_retval
+ setattr(rpc, rpc_method, rpc_method_mock)
+
+ retval = getattr(rpcapi, method)(ctxt, **kwargs)
+
+ self.assertEqual(retval, expected_retval)
+
+ expected_args = [ctxt, topic, expected_msg]
+ for arg, expected_arg in zip(rpc_method_mock.call_args[0],
+ expected_args):
+ self.assertEqual(arg, expected_arg)
+
+ def test_delete_network(self):
+ rpcapi = ana.AgentNotifierApi(topics.AGENT)
+ self._test_hyperv_quantum_api(
+ rpcapi,
+ topics.get_topic_name(
+ topics.AGENT,
+ topics.NETWORK,
+ topics.DELETE),
+ 'network_delete', rpc_method='fanout_cast',
+ network_id='fake_request_spec')
+
+ def test_port_update(self):
+ rpcapi = ana.AgentNotifierApi(topics.AGENT)
+ self._test_hyperv_quantum_api(
+ rpcapi,
+ topics.get_topic_name(
+ topics.AGENT,
+ topics.PORT,
+ topics.UPDATE),
+ 'port_update', rpc_method='fanout_cast',
+ port='fake_port',
+ network_type='fake_network_type',
+ segmentation_id='fake_segmentation_id',
+ physical_network='fake_physical_network')
+
+ def test_port_delete(self):
+ rpcapi = ana.AgentNotifierApi(topics.AGENT)
+ self._test_hyperv_quantum_api(
+ rpcapi,
+ topics.get_topic_name(
+ topics.AGENT,
+ topics.PORT,
+ topics.DELETE),
+ 'port_delete', rpc_method='fanout_cast',
+ port_id='port_id')
+
+ def test_tunnel_update(self):
+ rpcapi = ana.AgentNotifierApi(topics.AGENT)
+ self._test_hyperv_quantum_api(
+ rpcapi,
+ topics.get_topic_name(
+ topics.AGENT,
+ constants.TUNNEL,
+ topics.UPDATE),
+ 'tunnel_update', rpc_method='fanout_cast',
+ tunnel_ip='fake_ip', tunnel_id='fake_id')
+
+ def test_device_details(self):
+ rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
+ self._test_hyperv_quantum_api(
+ rpcapi, topics.PLUGIN,
+ 'get_device_details', rpc_method='call',
+ device='fake_device',
+ agent_id='fake_agent_id')
+
+ def test_update_device_down(self):
+ rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
+ self._test_hyperv_quantum_api(
+ rpcapi, topics.PLUGIN,
+ 'update_device_down', rpc_method='call',
+ device='fake_device',
+ agent_id='fake_agent_id')
+
+ def test_tunnel_sync(self):
+ rpcapi = agent_rpc.PluginApi(topics.PLUGIN)
+ self._test_hyperv_quantum_api(
+ rpcapi, topics.PLUGIN,
+ 'tunnel_sync', rpc_method='call',
+ tunnel_ip='fake_tunnel_ip')
Code Review / openstack-build / neutron-build.git / commitdiff