openstack
/
python-neutronclient
Code Issues Proposed changes

SFC plugin: fix port chain set commands 

Setting a flow classifier or port pair group currently fails,
additionally with multiple flow classifiers only the last one is added

* Fix adding multiple items in the same command
* Update unit test to try setting multiple flow classifiers

Change-Id: Ifbc24d84022c9893d3348fb27b20206fa464d8b2
Closes-Bug: #1716942
This commit is contained in:
Bernard Cafarelli 2017-10-17 12:18:18 +02:00
parent 0907ccc4df
commit 864f4d111f
No known key found for this signature in database
GPG Key ID: D148244A3C2462BD
2 changed files with 37 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
neutronclient/osc/v2/sfc/sfc_port_chain.py
View File

@ -188,37 +188,38 @@ class SetSfcPortChain(command.Command):
if parsed_args.no_flow_classifier:
attrs['flow_classifiers'] = []
if parsed_args.flow_classifiers:
for fc in parsed_args.flow_classifiers:
added = [client.find_resource(
'flow_classifier', fc,
cmd_resource='sfc_flow_classifier')['id']]
if parsed_args.no_flow_classifier:
existing = []
fc_list = []
else:
existing = [client.find_resource(
fc_list = client.find_resource(
resource, parsed_args.port_chain,
cmd_resource='sfc_port_chain')['flow_classifiers']]
attrs['flow_classifiers'] = sorted(list(
set(existing) | set(added)))
cmd_resource='sfc_port_chain')['flow_classifiers']
for fc in parsed_args.flow_classifiers:
fc_list.append(client.find_resource(
'flow_classifier', fc,
cmd_resource='sfc_flow_classifier')['id'])
attrs['flow_classifiers'] = sorted(list(set(fc_list)))
if (parsed_args.no_port_pair_group and not
parsed_args.port_pair_groups):
message = _('At least one --port-pair-group must be specified.')
raise exceptions.CommandError(message)
if parsed_args.no_port_pair_group and parsed_args.port_pair_groups:
ppg_list = []
for ppg in parsed_args.port_pair_groups:
attrs['port_pair_groups'] = [client.find_resource(
'port_pair_group', ppg,
cmd_resource='sfc_port_pair_group')['id']]
if (parsed_args.port_pair_groups and
not parsed_args.no_port_pair_group):
existing_ppg = [client.find_resource(
resource, parsed_args.port_chain,
cmd_resource='sfc_port_chain')['port_pair_groups']]
for ppg in parsed_args.port_pair_groups:
existing_ppg.append(client.find_resource(
ppg_list.append(client.find_resource(
'port_pair_group', ppg,
cmd_resource='sfc_port_pair_group')['id'])
attrs['port_pair_groups'] = sorted(list(set(existing_ppg)))
attrs['port_pair_groups'] = sorted(list(set(ppg_list)))
if (parsed_args.port_pair_groups and
not parsed_args.no_port_pair_group):
ppg_list = client.find_resource(
resource, parsed_args.port_chain,
cmd_resource='sfc_port_chain')['port_pair_groups']
for ppg in parsed_args.port_pair_groups:
ppg_list.append(client.find_resource(
'port_pair_group', ppg,
cmd_resource='sfc_port_pair_group')['id'])
attrs['port_pair_groups'] = sorted(list(set(ppg_list)))
body = {resource: attrs}
try:
client.update_sfc_port_chain(pc_id, body)

25
neutronclient/tests/unit/osc/v2/sfc/test_port_chain.py
View File

@ -266,35 +266,42 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2):
attrs)
self.assertIsNone(result)
def test_set_flow_classifier(self):
def test_set_flow_classifiers(self):
target = self.resource['id']
fc1 = 'flow_classifier1'
fc2 = 'flow_classifier2'
def _mock_flow_classifier(*args, **kwargs):
if self.neutronclient.find_resource.call_count == 1:
self.neutronclient.find_resource.assert_called_with(
self.res, target, cmd_resource='sfc_port_chain')
return {'flow_classifiers': [self.pc_fc]}
if self.neutronclient.find_resource.call_count == 2:
self.neutronclient.find_resource.assert_called_with(
'flow_classifier', fc1, cmd_resource='sfc_flow_classifier')
return {'id': args[1]}
if self.neutronclient.find_resource.call_count == 2:
if self.neutronclient.find_resource.call_count == 3:
self.neutronclient.find_resource.assert_called_with(
self.res, target, cmd_resource='sfc_port_chain')
return {'flow_classifiers': self.pc_fc}
'flow_classifier', fc2, cmd_resource='sfc_flow_classifier')
return {'id': args[1]}
self.neutronclient.find_resource.side_effect = _mock_flow_classifier
arglist = [
target,
'--flow-classifier', fc1,
'--flow-classifier', fc2,
]
verifylist = [
(self.res, target),
('flow_classifiers', [fc1])
('flow_classifiers', [fc1, fc2])
]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
result = self.cmd.take_action(parsed_args)
expect = {'flow_classifiers': sorted([self.pc_fc, fc1])}
expect = {'flow_classifiers': sorted([self.pc_fc, fc1, fc2])}
self.mocked.assert_called_once_with(target, {self.res: expect})
self.assertEqual(2, self.neutronclient.find_resource.call_count)
self.assertEqual(3, self.neutronclient.find_resource.call_count)
self.assertIsNone(result)
def test_set_no_flow_classifier(self):
@ -325,7 +332,7 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2):
if self.neutronclient.find_resource.call_count == 1:
self.neutronclient.find_resource.assert_called_with(
self.res, target, cmd_resource='sfc_port_chain')
return {'port_pair_groups': self.pc_ppg}
return {'port_pair_groups': [self.pc_ppg]}
if self.neutronclient.find_resource.call_count == 2:
self.neutronclient.find_resource.assert_called_with(