openstack/neutron
Code Issues Proposed changes

Merge "Enable delete bound trunk for linux bridge agent"

Browse Source
This commit is contained in:
Zuul 2018-12-10 22:07:09 +00:00 committed by Gerrit Code Review
commit 81bc51cc29
4 changed files with 26 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
neutron/services/trunk/plugin.py
View File

@ -264,7 +264,7 @@ class TrunkPlugin(service_base.ServicePluginBase,
trunk = self._get_trunk(context, trunk_id)
rules.trunk_can_be_managed(context, trunk)
trunk_port_validator = rules.TrunkPortValidator(trunk.port_id)
if not trunk_port_validator.is_bound(context):
if trunk_port_validator.can_be_trunked_or_untrunked(context):
# NOTE(status_police): when a trunk is deleted, the logical
# object disappears from the datastore, therefore there is no
# status transition involved. If PRECOMMIT failures occur,

4
neutron/services/trunk/rules.py
View File

@ -91,7 +91,7 @@ class TrunkPortValidator(object):
# resources (namely it is bound). Bound ports may be used as
# trunk parents, but that depends on the underlying driver in
# charge.
if not self.can_be_trunked(context):
if not self.can_be_trunked_or_untrunked(context):
raise trunk_exc.ParentPortInUse(port_id=self.port_id)
else:
# if the port is being used as subport in a trunk, check if it is a
@ -110,7 +110,7 @@ class TrunkPortValidator(object):
self._port = core_plugin.get_port(context, self.port_id)
return bool(self._port.get(portbindings.HOST_ID))
def can_be_trunked(self, context):
def can_be_trunked_or_untrunked(self, context):
""""Return true if a port can be trunked."""
if not self.is_bound(context):
# An unbound port can be trunked, always.

11
neutron/tests/unit/services/trunk/test_plugin.py
View File

@ -27,6 +27,7 @@ from neutron.services.trunk import constants
from neutron.services.trunk import drivers
from neutron.services.trunk import exceptions as trunk_exc
from neutron.services.trunk import plugin as trunk_plugin
from neutron.services.trunk import rules
from neutron.services.trunk.seg_types import validators
from neutron.tests.unit.plugins.ml2 import test_plugin
from neutron.tests.unit.services.trunk import fakes
@ -93,11 +94,19 @@ class TrunkPluginTestCase(test_plugin.Ml2PluginV2TestCase):
def test_delete_trunk_raise_in_use(self):
with self.port() as port:
fakes.FakeDriverCanTrunkBoundPort.create()
self.trunk_plugin = trunk_plugin.TrunkPlugin()
directory.add_plugin('trunk', self.trunk_plugin)
trunk = self._create_test_trunk(port)
core_plugin = directory.get_plugin()
port['port']['binding:host_id'] = 'host'
core_plugin.update_port(self.context, port['port']['id'], port)
self.assertRaises(trunk_exc.TrunkInUse,
trunk_port_validator = rules.TrunkPortValidator(trunk['port_id'])
if not trunk_port_validator.can_be_trunked_or_untrunked(
self.context):
self.assertRaises(trunk_exc.TrunkInUse,
self.trunk_plugin.delete_trunk,
self.context, trunk['id'])

22
neutron/tests/unit/services/trunk/test_rules.py
View File

@ -280,12 +280,13 @@ class TrunkPortValidatorTestCase(test_plugin.Ml2PluginV2TestCase):
def test_validate_port_cannot_be_trunked_raises(self):
with self.port() as port, \
mock.patch.object(rules.TrunkPortValidator,
"can_be_trunked", return_value=False), \
"can_be_trunked_or_untrunked",
return_value=False), \
testtools.ExpectedException(trunk_exc.ParentPortInUse):
validator = rules.TrunkPortValidator(port['port']['id'])
validator.validate(self.context)
def test_can_be_trunked_returns_false(self):
def test_can_be_trunked_or_untrunked_returns_false(self):
# need to trigger a driver registration
fakes.FakeDriverCanTrunkBoundPort.create()
self.trunk_plugin = trunk_plugin.TrunkPlugin()
@ -296,9 +297,10 @@ class TrunkPortValidatorTestCase(test_plugin.Ml2PluginV2TestCase):
core_plugin.update_port(self.context, port['port']['id'], port)
validator = rules.TrunkPortValidator(port['port']['id'])
# port cannot be trunked because of binding mismatch
self.assertFalse(validator.can_be_trunked(self.context))
self.assertFalse(
validator.can_be_trunked_or_untrunked(self.context))
def test_can_be_trunked_returns_true(self):
def test_can_be_trunked_or_untrunked_returns_true(self):
# need to trigger a driver registration
fakes.FakeDriverCanTrunkBoundPort.create()
self.trunk_plugin = trunk_plugin.TrunkPlugin()
@ -310,15 +312,17 @@ class TrunkPortValidatorTestCase(test_plugin.Ml2PluginV2TestCase):
port['port']['binding:host_id'] = 'host'
core_plugin.update_port(self.context, port['port']['id'], port)
validator = rules.TrunkPortValidator(port['port']['id'])
self.assertTrue(validator.can_be_trunked(self.context))
self.assertTrue(
validator.can_be_trunked_or_untrunked(self.context))
self.assertTrue(g.call_count)
def test_can_be_trunked_unbound_port(self):
def test_can_be_trunked_or_untrunked_unbound_port(self):
with self.port() as port:
validator = rules.TrunkPortValidator(port['port']['id'])
self.assertTrue(validator.can_be_trunked(self.context))
self.assertTrue(
validator.can_be_trunked_or_untrunked(self.context))
def test_can_be_trunked_raises_conflict(self):
def test_can_be_trunked_or_untrunked_raises_conflict(self):
d1 = fakes.FakeDriver.create()
d2 = fakes.FakeDriverWithAgent.create()
self.trunk_plugin = trunk_plugin.TrunkPlugin()
@ -333,7 +337,7 @@ class TrunkPortValidatorTestCase(test_plugin.Ml2PluginV2TestCase):
validator = rules.TrunkPortValidator(port['port']['id'])
self.assertRaises(
trunk_exc.TrunkPluginDriverConflict,
validator.can_be_trunked, self.context)
validator.can_be_trunked_or_untrunked, self.context)
def test_check_not_in_use_pass(self):
with self.port() as port: