openstack
/
neutron
Code Issues Proposed changes

l2pop UT: Simplify migration tests 

"port2" is created but not used in the tests.

Change-Id: Ib27d32063a2b5cecc707a6aece4e604cbfecefa7
This commit is contained in:
YAMAMOTO Takashi 2015-02-23 13:23:53 +09:00
parent a6af531339
commit 4791746f41
1 changed files with 66 additions and 88 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

154
neutron/tests/unit/plugins/ml2/drivers/l2pop/test_mech_driver.py
View File

@ -800,109 +800,87 @@ class TestL2PopulationRpcTestCase(test_plugin.Ml2PluginV2TestCase):
self._register_ml2_agents() self._register_ml2_agents()
with self.subnet(network=self._network) as subnet: with self.subnet(network=self._network) as subnet:
host_arg = {portbindings.HOST_ID: L2_AGENT['host']} host_arg = {portbindings.HOST_ID: L2_AGENT['host']}
host2_arg = {portbindings.HOST_ID: L2_AGENT_2['host']}
with self.port(subnet=subnet, cidr='10.0.0.0/24', with self.port(subnet=subnet, cidr='10.0.0.0/24',
device_owner=DEVICE_OWNER_COMPUTE, device_owner=DEVICE_OWNER_COMPUTE,
arg_list=(portbindings.HOST_ID,), arg_list=(portbindings.HOST_ID,),
**host_arg) as port1: **host_arg) as port1:
with self.port(subnet=subnet, cidr='10.0.0.0/24', p1 = port1['port']
device_owner=DEVICE_OWNER_COMPUTE, device1 = 'tap' + p1['id']
arg_list=(portbindings.HOST_ID,), self.callbacks.update_device_up(
**host2_arg) as port2: self.adminContext,
p1 = port1['port'] agent_id=L2_AGENT['host'],
device1 = 'tap' + p1['id'] device=device1)
self.callbacks.update_device_up( data2 = {'port': {'binding:host_id': L2_AGENT_2['host']}}
self.adminContext, req = self.new_update_request('ports', data2, p1['id'])
agent_id=L2_AGENT['host'], res = self.deserialize(self.fmt,
device=device1) req.get_response(self.api))
p2 = port2['port'] self.assertEqual(res['port']['binding:host_id'],
device2 = 'tap' + p2['id'] L2_AGENT_2['host'])
self.callbacks.update_device_up( self.mock_fanout.reset_mock()
self.adminContext, # NOTE(yamamoto): see bug #1441488
agent_id=L2_AGENT_2['host'], self.adminContext.session.expire_all()
device=device2) self.callbacks.get_device_details(
data2 = {'port': {'binding:host_id': L2_AGENT_2['host']}} self.adminContext,
req = self.new_update_request('ports', data2, p1['id']) device=device1,
res = self.deserialize(self.fmt, agent_id=L2_AGENT_2['host'])
req.get_response(self.api)) p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
self.assertEqual(res['port']['binding:host_id'], expected = {p1['network_id']:
L2_AGENT_2['host']) {'ports':
self.mock_fanout.reset_mock() {'20.0.0.1': [constants.FLOODING_ENTRY,
# NOTE(yamamoto): see bug #1441488 l2pop_rpc.PortInfo(
self.adminContext.session.expire_all() p1['mac_address'],
self.callbacks.get_device_details( p1_ips[0])]},
self.adminContext, 'network_type': 'vxlan',
device=device1, 'segment_id': 1}}
agent_id=L2_AGENT_2['host'])
p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
expected = {p1['network_id']:
{'ports':
{'20.0.0.1': [constants.FLOODING_ENTRY,
l2pop_rpc.PortInfo(
p1['mac_address'],
p1_ips[0])]},
'network_type': 'vxlan',
'segment_id': 1}}
self.mock_fanout.assert_called_with( self.mock_fanout.assert_called_with(
mock.ANY, 'remove_fdb_entries', expected) mock.ANY, 'remove_fdb_entries', expected)
def test_host_changed_twice(self): def test_host_changed_twice(self):
self._register_ml2_agents() self._register_ml2_agents()
with self.subnet(network=self._network) as subnet: with self.subnet(network=self._network) as subnet:
host_arg = {portbindings.HOST_ID: L2_AGENT['host']} host_arg = {portbindings.HOST_ID: L2_AGENT['host']}
host2_arg = {portbindings.HOST_ID: L2_AGENT_2['host']}
with self.port(subnet=subnet, cidr='10.0.0.0/24', with self.port(subnet=subnet, cidr='10.0.0.0/24',
device_owner=DEVICE_OWNER_COMPUTE, device_owner=DEVICE_OWNER_COMPUTE,
arg_list=(portbindings.HOST_ID,), arg_list=(portbindings.HOST_ID,),
**host_arg) as port1: **host_arg) as port1:
with self.port(subnet=subnet, cidr='10.0.0.0/24', p1 = port1['port']
device_owner=DEVICE_OWNER_COMPUTE, device1 = 'tap' + p1['id']
arg_list=(portbindings.HOST_ID,), self.callbacks.update_device_up(
**host2_arg) as port2: self.adminContext,
p1 = port1['port'] agent_id=L2_AGENT['host'],
device1 = 'tap' + p1['id'] device=device1)
self.callbacks.update_device_up( data2 = {'port': {'binding:host_id': L2_AGENT_2['host']}}
self.adminContext, req = self.new_update_request('ports', data2, p1['id'])
agent_id=L2_AGENT['host'], res = self.deserialize(self.fmt,
device=device1) req.get_response(self.api))
p2 = port2['port'] self.assertEqual(res['port']['binding:host_id'],
device2 = 'tap' + p2['id'] L2_AGENT_2['host'])
self.callbacks.update_device_up( data4 = {'port': {'binding:host_id': L2_AGENT_4['host']}}
self.adminContext, req = self.new_update_request('ports', data4, p1['id'])
agent_id=L2_AGENT_2['host'], res = self.deserialize(self.fmt,
device=device2) req.get_response(self.api))
data2 = {'port': {'binding:host_id': L2_AGENT_2['host']}} self.assertEqual(res['port']['binding:host_id'],
req = self.new_update_request('ports', data2, p1['id']) L2_AGENT_4['host'])
res = self.deserialize(self.fmt, self.mock_fanout.reset_mock()
req.get_response(self.api)) # NOTE(yamamoto): see bug #1441488
self.assertEqual(res['port']['binding:host_id'], self.adminContext.session.expire_all()
L2_AGENT_2['host']) self.callbacks.get_device_details(
data4 = {'port': {'binding:host_id': L2_AGENT_4['host']}} self.adminContext,
req = self.new_update_request('ports', data4, p1['id']) device=device1,
res = self.deserialize(self.fmt, agent_id=L2_AGENT_4['host'])
req.get_response(self.api)) p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
self.assertEqual(res['port']['binding:host_id'], expected = {p1['network_id']:
L2_AGENT_4['host']) {'ports':
self.mock_fanout.reset_mock() {'20.0.0.1': [constants.FLOODING_ENTRY,
# NOTE(yamamoto): see bug #1441488 l2pop_rpc.PortInfo(
self.adminContext.session.expire_all() p1['mac_address'],
self.callbacks.get_device_details( p1_ips[0])]},
self.adminContext, 'network_type': 'vxlan',
device=device1, 'segment_id': 1}}
agent_id=L2_AGENT_4['host'])
p1_ips = [p['ip_address'] for p in p1['fixed_ips']]
expected = {p1['network_id']:
{'ports':
{'20.0.0.1': [constants.FLOODING_ENTRY,
l2pop_rpc.PortInfo(
p1['mac_address'],
p1_ips[0])]},
'network_type': 'vxlan',
'segment_id': 1}}
self.mock_fanout.assert_called_with( self.mock_fanout.assert_called_with(
mock.ANY, 'remove_fdb_entries', expected) mock.ANY, 'remove_fdb_entries', expected)
def test_delete_port_invokes_update_device_down(self): def test_delete_port_invokes_update_device_down(self):
l2pop_mech = l2pop_mech_driver.L2populationMechanismDriver() l2pop_mech = l2pop_mech_driver.L2populationMechanismDriver()