x/vmware-nsx
Code Issues Proposed changes

Merge "Make sure we call BaseTestCase.setUp() first"

Browse Source
This commit is contained in:
Jenkins 2014-06-18 16:08:41 +00:00 committed by Gerrit Code Review
commit 86e2b1c4e6
2 changed files with 62 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

126
neutron/tests/unit/openvswitch/test_agent_scheduler.py
View File

@ -200,7 +200,7 @@ class OvsAgentSchedulerTestCaseBase(test_l3_plugin.L3NatTestCaseMixin,
l3_plugin = None l3_plugin = None
def setUp(self): def setUp(self):
# Save the global RESOURCE_ATTRIBUTE_MAP # Save the global RESOURCE_ATTRIBUTE_MAP before loading plugin
self.saved_attr_map = {} self.saved_attr_map = {}
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems(): for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy() self.saved_attr_map[resource] = attrs.copy()
@ -1020,17 +1020,16 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
'ovs_neutron_plugin.OVSNeutronPluginV2') 'ovs_neutron_plugin.OVSNeutronPluginV2')
def setUp(self): def setUp(self):
self.dhcp_notifier = dhcp_rpc_agent_api.DhcpAgentNotifyAPI() # Save the global RESOURCE_ATTRIBUTE_MAP before loading plugin
self.dhcp_notifier_cls_p = mock.patch(
'neutron.api.rpc.agentnotifiers.dhcp_rpc_agent_api.'
'DhcpAgentNotifyAPI')
self.dhcp_notifier_cls = self.dhcp_notifier_cls_p.start()
self.dhcp_notifier_cls.return_value = self.dhcp_notifier
# Save the global RESOURCE_ATTRIBUTE_MAP
self.saved_attr_map = {} self.saved_attr_map = {}
for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems(): for resource, attrs in attributes.RESOURCE_ATTRIBUTE_MAP.iteritems():
self.saved_attr_map[resource] = attrs.copy() self.saved_attr_map[resource] = attrs.copy()
super(OvsDhcpAgentNotifierTestCase, self).setUp(self.plugin_str) super(OvsDhcpAgentNotifierTestCase, self).setUp(self.plugin_str)
# the notifier is used to get access to make_msg() method only
self.dhcp_notifier = dhcp_rpc_agent_api.DhcpAgentNotifyAPI()
self.dhcp_notifier_cast = mock.patch(
'neutron.api.rpc.agentnotifiers.dhcp_rpc_agent_api.'
'DhcpAgentNotifyAPI.cast').start()
ext_mgr = extensions.PluginAwareExtensionManager.get_instance() ext_mgr = extensions.PluginAwareExtensionManager.get_instance()
self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr) self.ext_api = test_extensions.setup_extensions_middleware(ext_mgr)
self.adminContext = context.get_admin_context() self.adminContext = context.get_admin_context()
@ -1047,20 +1046,19 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map attributes.RESOURCE_ATTRIBUTE_MAP = self.saved_attr_map
def test_network_add_to_dhcp_agent_notification(self): def test_network_add_to_dhcp_agent_notification(self):
with mock.patch.object(self.dhcp_notifier, 'cast') as mock_dhcp: with self.network() as net1:
with self.network() as net1: network_id = net1['network']['id']
network_id = net1['network']['id'] self._register_agent_states()
self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP,
hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP, DHCP_HOSTA)
DHCP_HOSTA) self._add_network_to_dhcp_agent(hosta_id,
self._add_network_to_dhcp_agent(hosta_id, network_id)
network_id) self.dhcp_notifier_cast.assert_called_with(
mock_dhcp.assert_called_with( mock.ANY,
mock.ANY, self.dhcp_notifier.make_msg(
self.dhcp_notifier.make_msg( 'network_create_end',
'network_create_end', payload={'network': {'id': network_id}}),
payload={'network': {'id': network_id}}), topic='dhcp_agent.' + DHCP_HOSTA)
topic='dhcp_agent.' + DHCP_HOSTA)
def test_network_remove_from_dhcp_agent_notification(self): def test_network_remove_from_dhcp_agent_notification(self):
with self.network(do_delete=False) as net1: with self.network(do_delete=False) as net1:
@ -1070,27 +1068,27 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
DHCP_HOSTA) DHCP_HOSTA)
self._add_network_to_dhcp_agent(hosta_id, self._add_network_to_dhcp_agent(hosta_id,
network_id) network_id)
with mock.patch.object(self.dhcp_notifier, 'cast') as mock_dhcp:
self._remove_network_from_dhcp_agent(hosta_id, self._remove_network_from_dhcp_agent(hosta_id,
network_id) network_id)
mock_dhcp.assert_called_with( self.dhcp_notifier_cast.assert_called_with(
mock.ANY, mock.ANY,
self.dhcp_notifier.make_msg( self.dhcp_notifier.make_msg(
'network_delete_end', 'network_delete_end',
payload={'network_id': network_id}), payload={'network_id': network_id}),
topic='dhcp_agent.' + DHCP_HOSTA) topic='dhcp_agent.' + DHCP_HOSTA)
def test_agent_updated_dhcp_agent_notification(self): def test_agent_updated_dhcp_agent_notification(self):
with mock.patch.object(self.dhcp_notifier, 'cast') as mock_dhcp: self._register_agent_states()
self._register_agent_states() hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP,
hosta_id = self._get_agent_id(constants.AGENT_TYPE_DHCP, DHCP_HOSTA)
DHCP_HOSTA) self._disable_agent(hosta_id, admin_state_up=False)
self._disable_agent(hosta_id, admin_state_up=False)
mock_dhcp.assert_called_with( self.dhcp_notifier_cast.assert_called_with(
mock.ANY, self.dhcp_notifier.make_msg( mock.ANY, self.dhcp_notifier.make_msg(
'agent_updated', 'agent_updated',
payload={'admin_state_up': False}), payload={'admin_state_up': False}),
topic='dhcp_agent.' + DHCP_HOSTA) topic='dhcp_agent.' + DHCP_HOSTA)
def _network_port_create( def _network_port_create(
self, hosts, gateway=attributes.ATTR_NOT_SPECIFIED, owner=None): self, hosts, gateway=attributes.ATTR_NOT_SPECIFIED, owner=None):
@ -1102,22 +1100,21 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
'configurations': {'dhcp_driver': 'dhcp_driver', 'configurations': {'dhcp_driver': 'dhcp_driver',
'use_namespaces': True, }, 'use_namespaces': True, },
'agent_type': constants.AGENT_TYPE_DHCP}) 'agent_type': constants.AGENT_TYPE_DHCP})
with mock.patch.object(self.dhcp_notifier, 'cast') as mock_dhcp: with self.network(do_delete=False) as net1:
with self.network(do_delete=False) as net1: with self.subnet(network=net1,
with self.subnet(network=net1, gateway_ip=gateway,
gateway_ip=gateway, do_delete=False) as subnet1:
do_delete=False) as subnet1: if owner:
if owner: with self.port(subnet=subnet1,
with self.port(subnet=subnet1, no_delete=True,
no_delete=True, device_owner=owner) as port:
device_owner=owner) as port: return [net1, subnet1, port]
return [mock_dhcp, net1, subnet1, port] else:
else: with self.port(subnet=subnet1,
with self.port(subnet=subnet1, no_delete=True) as port:
no_delete=True) as port: return [net1, subnet1, port]
return [mock_dhcp, net1, subnet1, port]
def _notification_mocks(self, hosts, mock_dhcp, net, subnet, port): def _notification_mocks(self, hosts, net, subnet, port):
host_calls = {} host_calls = {}
for host in hosts: for host in hosts:
expected_calls = [ expected_calls = [
@ -1138,21 +1135,20 @@ class OvsDhcpAgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,
def test_network_port_create_notification(self): def test_network_port_create_notification(self):
hosts = [DHCP_HOSTA] hosts = [DHCP_HOSTA]
[mock_dhcp, net, subnet, port] = self._network_port_create(hosts) net, subnet, port = self._network_port_create(hosts)
expected_calls = self._notification_mocks(hosts, mock_dhcp, expected_calls = self._notification_mocks(hosts, net, subnet, port)
net, subnet, port) self.assertEqual(
self.assertEqual(expected_calls[DHCP_HOSTA], mock_dhcp.call_args_list) expected_calls[DHCP_HOSTA], self.dhcp_notifier_cast.call_args_list)
def test_network_ha_port_create_notification(self): def test_network_ha_port_create_notification(self):
cfg.CONF.set_override('dhcp_agents_per_network', 2) cfg.CONF.set_override('dhcp_agents_per_network', 2)
hosts = [DHCP_HOSTA, DHCP_HOSTC] hosts = [DHCP_HOSTA, DHCP_HOSTC]
[mock_dhcp, net, subnet, port] = self._network_port_create(hosts) net, subnet, port = self._network_port_create(hosts)
expected_calls = self._notification_mocks(hosts, mock_dhcp, expected_calls = self._notification_mocks(hosts, net, subnet, port)
net, subnet, port)
for expected in expected_calls[DHCP_HOSTA]: for expected in expected_calls[DHCP_HOSTA]:
self.assertIn(expected, mock_dhcp.call_args_list) self.assertIn(expected, self.dhcp_notifier_cast.call_args_list)
for expected in expected_calls[DHCP_HOSTC]: for expected in expected_calls[DHCP_HOSTC]:
self.assertIn(expected, mock_dhcp.call_args_list) self.assertIn(expected, self.dhcp_notifier_cast.call_args_list)
class OvsL3AgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin, class OvsL3AgentNotifierTestCase(test_l3_plugin.L3NatTestCaseMixin,

2
neutron/tests/unit/vmware/test_agent_scheduler.py
View File

@ -52,7 +52,7 @@ class DhcpAgentNotifierTestCase(test_base.OvsDhcpAgentNotifierTestCase):
cfg.CONF.set_override('metadata_mode', 'dhcp_host_route', 'NSX') cfg.CONF.set_override('metadata_mode', 'dhcp_host_route', 'NSX')
hosts = ['hosta'] hosts = ['hosta']
with mock.patch.object(rpc.LOG, 'info') as mock_log: with mock.patch.object(rpc.LOG, 'info') as mock_log:
[mock_dhcp, net, subnet, port] = self._network_port_create( net, subnet, port = self._network_port_create(
hosts, gateway=gateway, owner=constants.DEVICE_OWNER_DHCP) hosts, gateway=gateway, owner=constants.DEVICE_OWNER_DHCP)
self.assertEqual(subnet['subnet']['gateway_ip'], gateway) self.assertEqual(subnet['subnet']['gateway_ip'], gateway)
called = 1 if gateway is None else 0 called = 1 if gateway is None else 0