Browse Source

Merge "Support tagging Neutron ports on creation"

changes/33/751633/1
Zuul 1 year ago
committed by Gerrit Code Review
parent
commit
51aee432d9
  1. 1
      .zuul.yaml
  2. 15
      openstackclient/network/v2/port.py
  3. 58
      openstackclient/tests/unit/network/v2/test_port.py

1
.zuul.yaml

@ -91,6 +91,7 @@
neutron-segments: true
q-metering: true
q-qos: true
neutron-tag-ports-during-bulk-creation: true
tox_envlist: functional
- job:

15
openstackclient/network/v2/port.py

@ -480,12 +480,23 @@ class CreatePort(command.ShowOne):
if parsed_args.qos_policy:
attrs['qos_policy_id'] = client.find_qos_policy(
parsed_args.qos_policy, ignore_missing=False).id
set_tags_in_post = bool(
client.find_extension('tag-ports-during-bulk-creation'))
if set_tags_in_post:
if parsed_args.no_tag:
attrs['tags'] = []
if parsed_args.tags:
attrs['tags'] = list(set(parsed_args.tags))
with common.check_missing_extension_if_error(
self.app.client_manager.network, attrs):
obj = client.create_port(**attrs)
# tags cannot be set when created, so tags need to be set later.
_tag.update_tags_for_set(client, obj, parsed_args)
if not set_tags_in_post:
# tags cannot be set when created, so tags need to be set later.
_tag.update_tags_for_set(client, obj, parsed_args)
display_columns, columns = _get_columns(obj)
data = utils.get_item_properties(obj, columns, formatters=_formatters)

58
openstackclient/tests/unit/network/v2/test_port.py

@ -121,6 +121,7 @@ class TestCreatePort(TestPort):
self.network.find_network = mock.Mock(return_value=fake_net)
self.fake_subnet = network_fakes.FakeSubnet.create_one_subnet()
self.network.find_subnet = mock.Mock(return_value=self.fake_subnet)
self.network.find_extension = mock.Mock(return_value=[])
# Get the command object to test
self.cmd = port.CreatePort(self.app, self.namespace)
@ -536,7 +537,7 @@ class TestCreatePort(TestPort):
'name': 'test-port',
})
def _test_create_with_tag(self, add_tags=True):
def _test_create_with_tag(self, add_tags=True, add_tags_in_post=True):
arglist = [
'--network', self._port.network_id,
'test-port',
@ -555,28 +556,59 @@ class TestCreatePort(TestPort):
else:
verifylist.append(('no_tag', True))
self.network.find_extension = mock.Mock(return_value=add_tags_in_post)
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.network.create_port.assert_called_once_with(
admin_state_up=True,
network_id=self._port.network_id,
name='test-port'
)
if add_tags:
self.network.set_tags.assert_called_once_with(
self._port,
tests_utils.CompareBySet(['red', 'blue']))
args = {
'admin_state_up': True,
'network_id': self._port.network_id,
'name': 'test-port',
}
if add_tags_in_post:
if add_tags:
args['tags'] = sorted(['red', 'blue'])
else:
args['tags'] = []
self.network.create_port.assert_called_once()
# Now we need to verify if arguments to call create_port are as
# expected,
# But we can't simply use assert_called_once_with() method because
# duplicates from 'tags' are removed with
# list(set(parsed_args.tags)) and that don't quarantee order of
# tags list which is used to call create_port().
create_port_call_kwargs = self.network.create_port.call_args[1]
create_port_call_kwargs['tags'] = sorted(
create_port_call_kwargs['tags'])
self.assertDictEqual(args, create_port_call_kwargs)
else:
self.assertFalse(self.network.set_tags.called)
self.network.create_port.assert_called_once_with(
admin_state_up=True,
network_id=self._port.network_id,
name='test-port'
)
if add_tags:
self.network.set_tags.assert_called_once_with(
self._port,
tests_utils.CompareBySet(['red', 'blue']))
else:
self.assertFalse(self.network.set_tags.called)
self.assertEqual(self.columns, columns)
self.assertItemEqual(self.data, data)
def test_create_with_tags(self):
self._test_create_with_tag(add_tags=True)
self._test_create_with_tag(add_tags=True, add_tags_in_post=True)
def test_create_with_no_tag(self):
self._test_create_with_tag(add_tags=False)
self._test_create_with_tag(add_tags=False, add_tags_in_post=True)
def test_create_with_tags_using_put(self):
self._test_create_with_tag(add_tags=True, add_tags_in_post=False)
def test_create_with_no_tag_using_put(self):
self._test_create_with_tag(add_tags=False, add_tags_in_post=False)
def _test_create_with_uplink_status_propagation(self, enable=True):
arglist = [

Loading…
Cancel
Save