unbreak unit test caused by c5fa665de3

The change set of c5fa665de3
introduced a new test TestMl2PortsV2.test_update_port_host_id_changed
which isn't always correct depending on driver.
ML2 driver may change port status to ACTIVE on port binding.
With such driver, the test fails. Grep showed 5 decomposed driver
don't have the assumption.
So this patch makes the test to run only _process_bind_port() without
calling full update_port() so that precommit/postcommit don't get
involved.

Change-Id: Ib034e2121914e5f253eb673261f26b4c8487f431
Closes-Bug: #1545218
(cherry picked from commit ad7b72963d)
This commit is contained in:
Isaku Yamahata 2016-02-12 18:53:38 -08:00 committed by Isaku Yamahata
parent 2580cf7dd2
commit fa2c3784a6
2 changed files with 34 additions and 10 deletions

View File

@ -496,16 +496,6 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase):
plugin.update_port(ctx, port['port']['id'], port) plugin.update_port(ctx, port['port']['id'], port)
self.assertTrue(sg_member_update.called) self.assertTrue(sg_member_update.called)
def test_update_port_host_id_changed(self):
ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin()
host_id = {portbindings.HOST_ID: 'host1'}
with self.port(**host_id) as port:
plugin.update_port_status(ctx, port['port']['id'], 'UP')
port['port']['binding:host_id'] = 'host2'
result = plugin.update_port(ctx, port['port']['id'], port)
self.assertEqual(constants.PORT_STATUS_DOWN, result['status'])
def test_update_port_status_with_network(self): def test_update_port_status_with_network(self):
ctx = context.get_admin_context() ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin() plugin = manager.NeutronManager.get_plugin()

View File

@ -20,6 +20,7 @@ from neutron import context
from neutron.extensions import portbindings from neutron.extensions import portbindings
from neutron import manager from neutron import manager
from neutron.plugins.ml2 import config as config from neutron.plugins.ml2 import config as config
from neutron.plugins.ml2 import driver_context
from neutron.plugins.ml2 import models as ml2_models from neutron.plugins.ml2 import models as ml2_models
from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin from neutron.tests.unit.db import test_db_base_plugin_v2 as test_plugin
@ -178,6 +179,39 @@ class PortBindingTestCase(test_plugin.NeutronDbPluginV2TestCase):
def test_update_from_host_to_empty_binding_notifies_agent(self): def test_update_from_host_to_empty_binding_notifies_agent(self):
self._test_update_port_binding('host-ovs-no_filter', '') self._test_update_port_binding('host-ovs-no_filter', '')
def test_process_binding_port_host_id_changed(self):
ctx = context.get_admin_context()
plugin = manager.NeutronManager.get_plugin()
host_id = {portbindings.HOST_ID: 'host1'}
with self.port(**host_id) as port:
# Since the port is DOWN at first
# It's necessary to make its status ACTIVE for this test
plugin.update_port_status(ctx, port['port']['id'],
const.PORT_STATUS_ACTIVE)
attrs = port['port']
attrs['status'] = const.PORT_STATUS_ACTIVE
original_port = attrs.copy()
attrs['binding:host_id'] = 'host2'
updated_port = attrs.copy()
network = {'id': attrs['network_id']}
binding = ml2_models.PortBinding(
port_id=original_port['id'],
host=original_port['binding:host_id'],
vnic_type=original_port['binding:vnic_type'],
profile=original_port['binding:profile'],
vif_type=original_port['binding:vif_type'],
vif_details=original_port['binding:vif_details'])
levels = 1
mech_context = driver_context.PortContext(
plugin, ctx, updated_port, network, binding, levels,
original_port=original_port)
plugin._process_port_binding(mech_context, port['port'])
self.assertEqual(const.PORT_STATUS_DOWN, updated_port['status'])
port_dict = plugin.get_port(ctx, port['port']['id'])
self.assertEqual(const.PORT_STATUS_DOWN, port_dict['status'])
def test_dvr_binding(self): def test_dvr_binding(self):
ctx = context.get_admin_context() ctx = context.get_admin_context()
with self.port(device_owner=const.DEVICE_OWNER_DVR_INTERFACE) as port: with self.port(device_owner=const.DEVICE_OWNER_DVR_INTERFACE) as port: