From fa2c3784a65f293ff44c1d3dac3f91065f0dfc97 Mon Sep 17 00:00:00 2001 From: Isaku Yamahata Date: Fri, 12 Feb 2016 18:53:38 -0800 Subject: [PATCH] unbreak unit test caused by c5fa665de3173f3ad82cc3e7624b5968bc52c08d The change set of c5fa665de3173f3ad82cc3e7624b5968bc52c08d 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 ad7b72963d462f9a512eca3b5a054d8ea33f2324) --- neutron/tests/unit/plugins/ml2/test_plugin.py | 10 ------ .../unit/plugins/ml2/test_port_binding.py | 34 +++++++++++++++++++ 2 files changed, 34 insertions(+), 10 deletions(-) diff --git a/neutron/tests/unit/plugins/ml2/test_plugin.py b/neutron/tests/unit/plugins/ml2/test_plugin.py index 26233a32610..03c0b409361 100644 --- a/neutron/tests/unit/plugins/ml2/test_plugin.py +++ b/neutron/tests/unit/plugins/ml2/test_plugin.py @@ -496,16 +496,6 @@ class TestMl2PortsV2(test_plugin.TestPortsV2, Ml2PluginV2TestCase): plugin.update_port(ctx, port['port']['id'], port) 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): ctx = context.get_admin_context() plugin = manager.NeutronManager.get_plugin() diff --git a/neutron/tests/unit/plugins/ml2/test_port_binding.py b/neutron/tests/unit/plugins/ml2/test_port_binding.py index 3f7321859e4..3cb6352f267 100644 --- a/neutron/tests/unit/plugins/ml2/test_port_binding.py +++ b/neutron/tests/unit/plugins/ml2/test_port_binding.py @@ -20,6 +20,7 @@ from neutron import context from neutron.extensions import portbindings from neutron import manager 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.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): 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): ctx = context.get_admin_context() with self.port(device_owner=const.DEVICE_OWNER_DVR_INTERFACE) as port: