openstack
/
python-neutronclient
Code Issues Proposed changes

SFC plugin: fix all list commands 

Rename all list_ functions to follow the expected naming format
expected by find_resource() calls: "list_%s" % cmd_resource_plural

Rename for consistency the other SFC commands to include the "_sfc_"
prefix

Change-Id: Ie01692d5cbe928b042c7e8052295a781ac205764
Related-Bug: #1716942
This commit is contained in:
Bernard Cafarelli 2017-10-17 12:14:59 +02:00
parent 8352c391d3
commit 0907ccc4df
No known key found for this signature in database
GPG Key ID: D148244A3C2462BD
9 changed files with 106 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
neutronclient/osc/v2/sfc/sfc_flow_classifier.py
View File

@ -117,7 +117,7 @@ class CreateSfcFlowClassifier(command.ShowOne):
client = self.app.client_manager.neutronclient
attrs = _get_common_attrs(self.app.client_manager, parsed_args)
body = {resource: attrs}
obj = client.create_flow_classifier(body)[resource]
obj = client.create_sfc_flow_classifier(body)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -140,7 +140,7 @@ class DeleteSfcFlowClassifier(command.Command):
client = self.app.client_manager.neutronclient
fc_id = _get_id(client, parsed_args.flow_classifier, resource)
try:
client.delete_flow_classifier(fc_id)
client.delete_sfc_flow_classifier(fc_id)
except Exception as e:
msg = (_("Failed to delete flow classifier with name "
"or ID '%(fc)s': %(e)s")
@ -198,7 +198,7 @@ class ListSfcFlowClassifier(command.Lister):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
obj = client.list_flow_classifier()
obj = client.list_sfc_flow_classifiers()
obj_extend = self.extend_list(obj, parsed_args)
headers, columns = nc_osc_utils.get_column_definitions(
_attr_map, long_listing=parsed_args.long)
@ -233,7 +233,7 @@ class SetSfcFlowClassifier(command.Command):
is_create=False)
body = {resource: attrs}
try:
client.update_flow_classifier(fc_id, body)
client.update_sfc_flow_classifier(fc_id, body)
except Exception as e:
msg = (_("Failed to update flow classifier '%(fc)s': %(e)s")
% {'fc': parsed_args.flow_classifier, 'e': e})
@ -255,7 +255,7 @@ class ShowSfcFlowClassifier(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
fc_id = _get_id(client, parsed_args.flow_classifier, resource)
obj = client.show_flow_classifier(fc_id)[resource]
obj = client.show_sfc_flow_classifier(fc_id)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data

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

@ -83,7 +83,7 @@ class CreateSfcPortChain(command.ShowOne):
client = self.app.client_manager.neutronclient
attrs = _get_common_attrs(self.app.client_manager, parsed_args)
body = {resource: attrs}
obj = client.create_port_chain(body)[resource]
obj = client.create_sfc_port_chain(body)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -106,7 +106,7 @@ class DeleteSfcPortChain(command.Command):
client = self.app.client_manager.neutronclient
pc_id = _get_id(client, parsed_args.port_chain, resource)
try:
client.delete_port_chain(pc_id)
client.delete_sfc_port_chain(pc_id)
except Exception as e:
msg = (_("Failed to delete port chain with name "
"or ID '%(pc)s': %(e)s")
@ -129,7 +129,7 @@ class ListSfcPortChain(command.Lister):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
data = client.list_port_chain()
data = client.list_sfc_port_chains()
headers, columns = nc_osc_utils.get_column_definitions(
_attr_map, long_listing=parsed_args.long)
return (headers,
@ -221,7 +221,7 @@ class SetSfcPortChain(command.Command):
attrs['port_pair_groups'] = sorted(list(set(existing_ppg)))
body = {resource: attrs}
try:
client.update_port_chain(pc_id, body)
client.update_sfc_port_chain(pc_id, body)
except Exception as e:
msg = (_("Failed to update port chain '%(pc)s': %(e)s")
% {'pc': parsed_args.port_chain, 'e': e})
@ -243,7 +243,7 @@ class ShowSfcPortChain(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
pc_id = _get_id(client, parsed_args.port_chain, resource)
obj = client.show_port_chain(pc_id)[resource]
obj = client.show_sfc_port_chain(pc_id)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -310,7 +310,7 @@ class UnsetSfcPortChain(command.Command):
raise exceptions.CommandError(message)
body = {resource: attrs}
try:
client.update_port_chain(pc_id, body)
client.update_sfc_port_chain(pc_id, body)
except Exception as e:
msg = (_("Failed to unset port chain '%(pc)s': %(e)s")
% {'pc': parsed_args.port_chain, 'e': e})

10
neutronclient/osc/v2/sfc/sfc_port_pair.py
View File

@ -79,7 +79,7 @@ class CreateSfcPortPair(command.ShowOne):
client = self.app.client_manager.neutronclient
attrs = _get_common_attrs(self.app.client_manager, parsed_args)
body = {resource: attrs}
obj = client.create_port_pair(body)[resource]
obj = client.create_sfc_port_pair(body)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -102,7 +102,7 @@ class DeleteSfcPortPair(command.Command):
client = self.app.client_manager.neutronclient
port_pair_id = _get_id(client, parsed_args.port_pair, resource)
try:
client.delete_port_pair(port_pair_id)
client.delete_sfc_port_pair(port_pair_id)
except Exception as e:
msg = (_("Failed to delete port pair with name "
"or ID '%(port_pair)s': %(e)s")
@ -124,7 +124,7 @@ class ListSfcPortPair(command.Lister):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
data = client.list_port_pair()
data = client.list_sfc_port_pairs()
headers, columns = nc_osc_utils.get_column_definitions(
_attr_map, long_listing=parsed_args.long)
return (headers,
@ -160,7 +160,7 @@ class SetSfcPortPair(command.Command):
is_create=False)
body = {resource: attrs}
try:
client.update_port_pair(port_pair_id, body)
client.update_sfc_port_pair(port_pair_id, body)
except Exception as e:
msg = (_("Failed to update port pair '%(port_pair)s': %(e)s")
% {'port_pair': parsed_args.port_pair, 'e': e})
@ -182,7 +182,7 @@ class ShowSfcPortPair(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
port_pair_id = _get_id(client, parsed_args.port_pair, resource)
obj = client.show_port_pair(port_pair_id)[resource]
obj = client.show_sfc_port_pair(port_pair_id)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data

12
neutronclient/osc/v2/sfc/sfc_port_pair_group.py
View File

@ -74,7 +74,7 @@ class CreateSfcPortPairGroup(command.ShowOne):
client = self.app.client_manager.neutronclient
attrs = _get_common_attrs(self.app.client_manager, parsed_args)
body = {resource: attrs}
obj = client.create_port_pair_group(body)[resource]
obj = client.create_sfc_port_pair_group(body)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -97,7 +97,7 @@ class DeleteSfcPortPairGroup(command.Command):
client = self.app.client_manager.neutronclient
ppg_id = _get_id(client, parsed_args.port_pair_group, resource)
try:
client.delete_port_pair_group(ppg_id)
client.delete_sfc_port_pair_group(ppg_id)
except Exception as e:
msg = (_("Failed to delete port pair group with name "
"or ID '%(ppg)s': %(e)s")
@ -120,7 +120,7 @@ class ListSfcPortPairGroup(command.Lister):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
data = client.list_port_pair_group()
data = client.list_sfc_port_pair_groups()
headers, columns = nc_osc_utils.get_column_definitions(
_attr_map, long_listing=parsed_args.long)
return (headers,
@ -180,7 +180,7 @@ class SetSfcPortPairGroup(command.Command):
attrs['port_pairs'] = sorted(list(set(existing) | set(added)))
body = {resource: attrs}
try:
client.update_port_pair_group(ppg_id, body)
client.update_sfc_port_pair_group(ppg_id, body)
except Exception as e:
msg = (_("Failed to update port pair group '%(ppg)s': %(e)s")
% {'ppg': parsed_args.port_pair_group, 'e': e})
@ -202,7 +202,7 @@ class ShowSfcPortPairGroup(command.ShowOne):
def take_action(self, parsed_args):
client = self.app.client_manager.neutronclient
ppg_id = _get_id(client, parsed_args.port_pair_group, resource)
obj = client.show_port_pair_group(ppg_id)[resource]
obj = client.show_sfc_port_pair_group(ppg_id)[resource]
columns, display_columns = nc_osc_utils.get_columns(obj, _attr_map)
data = utils.get_dict_properties(obj, columns)
return display_columns, data
@ -247,7 +247,7 @@ class UnsetSfcPortPairGroup(command.Command):
attrs['port_pairs'] = []
body = {resource: attrs}
try:
client.update_port_pair_group(ppg_id, body)
client.update_sfc_port_pair_group(ppg_id, body)
except Exception as e:
msg = (_("Failed to unset port pair group '%(ppg)s': %(e)s")
% {'ppg': parsed_args.port_pair_group, 'e': e})

28
neutronclient/tests/unit/osc/v2/sfc/test_flow_classifier.py
View File

@ -67,7 +67,7 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestCreateSfcFlowClassifier, self).setUp()
mock.patch(get_id, new=_get_id).start()
self.neutronclient.create_flow_classifier = mock.Mock(
self.neutronclient.create_sfc_flow_classifier = mock.Mock(
return_value={'flow_classifier': self._fc})
self.data = self.get_data()
@ -90,7 +90,7 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_flow_classifier.assert_called_once_with({
self.neutronclient.create_sfc_flow_classifier.assert_called_once_with({
'flow_classifier': {
'name': self._fc['name'],
'logical_source_port': self._fc['logical_source_port'],
@ -129,7 +129,7 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_flow_classifier.assert_called_once_with({
self.neutronclient.create_sfc_flow_classifier.assert_called_once_with({
'flow_classifier': {
'name': self._fc['name'],
'description': self._fc['description'],
@ -155,14 +155,14 @@ class TestDeleteSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestDeleteSfcFlowClassifier, self).setUp()
mock.patch(get_id, new=_get_id).start()
self.neutronclient.delete_flow_classifier = mock.Mock(
self.neutronclient.delete_sfc_flow_classifier = mock.Mock(
return_value=None)
self.cmd = sfc_flow_classifier.DeleteSfcFlowClassifier(self.app,
self.namespace)
def test_delete_flow_classifier(self):
client = self.app.client_manager.neutronclient
mock_flow_classifier_delete = client.delete_flow_classifier
mock_flow_classifier_delete = client.delete_sfc_flow_classifier
arglist = [
self._flow_classifier[0]['id'],
]
@ -184,14 +184,14 @@ class TestSetSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestSetSfcFlowClassifier, self).setUp()
mock.patch(get_id, new=_get_id).start()
self.neutronclient.update_flow_classifier = mock.Mock(
self.neutronclient.update_sfc_flow_classifier = mock.Mock(
return_value=None)
self.cmd = sfc_flow_classifier.SetSfcFlowClassifier(self.app,
self.namespace)
def test_set_flow_classifier(self):
client = self.app.client_manager.neutronclient
mock_flow_classifier_update = client.update_flow_classifier
mock_flow_classifier_update = client.update_sfc_flow_classifier
arglist = [
self._flow_classifier_name,
'--name', 'name_updated',
@ -255,7 +255,7 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestShowSfcFlowClassifier, self).setUp()
mock.patch(get_id, new=_get_id).start()
self.neutronclient.show_flow_classifier = mock.Mock(
self.neutronclient.show_sfc_flow_classifier = mock.Mock(
return_value=self._flow_classifier
)
# Get the command object to test
@ -264,7 +264,7 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def test_show_flow_classifier(self):
client = self.app.client_manager.neutronclient
mock_flow_classifier_show = client.show_flow_classifier
mock_flow_classifier_show = client.show_sfc_flow_classifier
arglist = [
self._flow_classifier_id,
]
@ -325,19 +325,20 @@ class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def setUp(self):
super(TestListSfcFlowClassifier, self).setUp()
mock.patch(get_id, new=_get_id).start()
self.neutronclient.list_flow_classifier = mock.Mock(
self.neutronclient.list_sfc_flow_classifiers = mock.Mock(
return_value={'flow_classifiers': self._fc}
)
# Get the command object to test
self.cmd = sfc_flow_classifier.ListSfcFlowClassifier(self.app,
self.namespace)
def test_list_flow_classifier(self):
def test_list_flow_classifiers(self):
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns = self.cmd.take_action(parsed_args)
fcs = self.neutronclient.list_flow_classifier()['flow_classifiers']
fcs = self.neutronclient \
.list_sfc_flow_classifiers()['flow_classifiers']
fc = fcs[0]
data = [
fc['id'],
@ -354,7 +355,8 @@ class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2):
def test_list_with_long_option(self):
arglist = ['--long']
verifylist = [('long', True)]
fcs = self.neutronclient.list_flow_classifier()['flow_classifiers']
fcs = self.neutronclient \
.list_sfc_flow_classifiers()['flow_classifiers']
fc = fcs[0]
data = [
fc['id'],

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

@ -55,7 +55,7 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.neutronclient.create_port_chain = mock.Mock(
self.neutronclient.create_sfc_port_chain = mock.Mock(
return_value={'port_chain': self._port_chain})
self.data = self.get_data()
@ -77,7 +77,7 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_chain.assert_called_once_with({
self.neutronclient.create_sfc_port_chain.assert_called_once_with({
'port_chain': {
'name': self._port_chain['name'],
'port_pair_groups': [self._port_chain['port_pair_groups']],
@ -109,7 +109,7 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_chain.assert_called_once_with({
self.neutronclient.create_sfc_port_chain.assert_called_once_with({
'port_chain': {
'name': self._port_chain['name'],
'port_pair_groups': [self._port_chain['port_pair_groups']],
@ -131,12 +131,12 @@ class TestDeleteSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.neutronclient.delete_port_chain = mock.Mock(return_value=None)
self.neutronclient.delete_sfc_port_chain = mock.Mock(return_value=None)
self.cmd = sfc_port_chain.DeleteSfcPortChain(self.app, self.namespace)
def test_delete_port_chain(self):
client = self.app.client_manager.neutronclient
mock_port_chain_delete = client.delete_port_chain
mock_port_chain_delete = client.delete_sfc_port_chain
arglist = [
self._port_chain[0]['id'],
]
@ -183,18 +183,18 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.neutronclient.list_port_chain = mock.Mock(
self.neutronclient.list_sfc_port_chains = mock.Mock(
return_value={'port_chains': self._port_chains}
)
# Get the command object to test
self.cmd = sfc_port_chain.ListSfcPortChain(self.app, self.namespace)
def test_list_port_chain(self):
def test_list_port_chains(self):
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns = self.cmd.take_action(parsed_args)[0]
pcs = self.neutronclient.list_port_chain()['port_chains']
pcs = self.neutronclient.list_sfc_port_chains()['port_chains']
pc = pcs[0]
data = [
pc['id'],
@ -212,7 +212,7 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2):
verifylist = [('long', True)]
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns = self.cmd.take_action(parsed_args)[0]
pcs = self.neutronclient.list_port_chain()['port_chains']
pcs = self.neutronclient.list_sfc_port_chains()['port_chains']
pc = pcs[0]
data = [
pc['id'],
@ -242,12 +242,12 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.mocked = self.neutronclient.update_port_chain
self.mocked = self.neutronclient.update_sfc_port_chain
self.cmd = sfc_port_chain.SetSfcPortChain(self.app, self.namespace)
def test_set_port_chain(self):
client = self.app.client_manager.neutronclient
mock_port_chain_update = client.update_port_chain
mock_port_chain_update = client.update_sfc_port_chain
arglist = [
self._port_chain_name,
'--name', 'name_updated',
@ -299,7 +299,7 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2):
def test_set_no_flow_classifier(self):
client = self.app.client_manager.neutronclient
mock_port_chain_update = client.update_port_chain
mock_port_chain_update = client.update_sfc_port_chain
arglist = [
self._port_chain_name,
'--no-flow-classifier',
@ -429,7 +429,7 @@ class TestShowSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.neutronclient.show_port_chain = mock.Mock(
self.neutronclient.show_sfc_port_chain = mock.Mock(
return_value=self._port_chain
)
# Get the command object to test
@ -437,7 +437,7 @@ class TestShowSfcPortChain(fakes.TestNeutronClientOSCV2):
def test_show_port_chain(self):
client = self.app.client_manager.neutronclient
mock_port_chain_show = client.show_port_chain
mock_port_chain_show = client.show_sfc_port_chain
arglist = [
self._port_chain_id,
]
@ -466,9 +466,9 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_chain._get_id',
new=_get_id).start()
self.neutronclient.update_port_chain = mock.Mock(
self.neutronclient.update_sfc_port_chain = mock.Mock(
return_value=None)
self.mocked = self.neutronclient.update_port_chain
self.mocked = self.neutronclient.update_sfc_port_chain
self.cmd = sfc_port_chain.UnsetSfcPortChain(self.app, self.namespace)
def test_unset_port_pair_group(self):
@ -539,7 +539,7 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2):
def test_unset_all_flow_classifier(self):
client = self.app.client_manager.neutronclient
target = self.resource['id']
mock_port_chain_update = client.update_port_chain
mock_port_chain_update = client.update_sfc_port_chain
arglist = [
target,
'--all-flow-classifier',

26
neutronclient/tests/unit/osc/v2/sfc/test_port_pair.py
View File

@ -50,7 +50,7 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2):
super(TestCreateSfcPortPair, self).setUp()
mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id',
new=_get_id).start()
self.neutronclient.create_port_pair = mock.Mock(
self.neutronclient.create_sfc_port_pair = mock.Mock(
return_value={'port_pair': self._port_pair})
self.data = self.get_data()
@ -71,7 +71,7 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_pair.assert_called_once_with({
self.neutronclient.create_sfc_port_pair.assert_called_once_with({
'port_pair': {'name': self._port_pair['name'],
'ingress': self._port_pair['ingress'],
'egress': self._port_pair['egress'],
@ -104,7 +104,7 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_pair.assert_called_once_with({
self.neutronclient.create_sfc_port_pair.assert_called_once_with({
'port_pair': {'name': self._port_pair['name'],
'ingress': self._port_pair['ingress'],
'egress': self._port_pair['egress'],
@ -132,12 +132,12 @@ class TestDeleteSfcPortPair(fakes.TestNeutronClientOSCV2):
super(TestDeleteSfcPortPair, self).setUp()
mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id',
new=_get_id).start()
self.neutronclient.delete_port_pair = mock.Mock(return_value=None)
self.neutronclient.delete_sfc_port_pair = mock.Mock(return_value=None)
self.cmd = sfc_port_pair.DeleteSfcPortPair(self.app, self.namespace)
def test_delete_port_pair(self):
client = self.app.client_manager.neutronclient
mock_port_pair_delete = client.delete_port_pair
mock_port_pair_delete = client.delete_sfc_port_pair
arglist = [
self._port_pair[0]['id'],
]
@ -179,18 +179,18 @@ class TestListSfcPortPair(fakes.TestNeutronClientOSCV2):
super(TestListSfcPortPair, self).setUp()
mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id',
new=_get_id).start()
self.neutronclient.list_port_pair = mock.Mock(
self.neutronclient.list_sfc_port_pairs = mock.Mock(
return_value={'port_pairs': self._port_pairs}
)
# Get the command object to test
self.cmd = sfc_port_pair.ListSfcPortPair(self.app, self.namespace)
def test_list_port_pair(self):
def test_list_port_pairs(self):
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns = self.cmd.take_action(parsed_args)[0]
port_pairs = self.neutronclient.list_port_pair()['port_pairs']
port_pairs = self.neutronclient.list_sfc_port_pairs()['port_pairs']
port_pair = port_pairs[0]
data = [
port_pair['id'],
@ -204,7 +204,7 @@ class TestListSfcPortPair(fakes.TestNeutronClientOSCV2):
def test_list_with_long_option(self):
arglist = ['--long']
verifylist = [('long', True)]
port_pairs = self.neutronclient.list_port_pair()['port_pairs']
port_pairs = self.neutronclient.list_sfc_port_pairs()['port_pairs']
port_pair = port_pairs[0]
data = [
port_pair['id'],
@ -229,12 +229,12 @@ class TestSetSfcPortPair(fakes.TestNeutronClientOSCV2):
super(TestSetSfcPortPair, self).setUp()
mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id',
new=_get_id).start()
self.neutronclient.update_port_pair = mock.Mock(return_value=None)
self.neutronclient.update_sfc_port_pair = mock.Mock(return_value=None)
self.cmd = sfc_port_pair.SetSfcPortPair(self.app, self.namespace)
def test_set_port_pair(self):
client = self.app.client_manager.neutronclient
mock_port_pair_update = client.update_port_pair
mock_port_pair_update = client.update_sfc_port_pair
arglist = [
self._port_pair_name,
'--name', 'name_updated',
@ -286,7 +286,7 @@ class TestShowSfcPortPair(fakes.TestNeutronClientOSCV2):
mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id',
new=_get_id).start()
self.neutronclient.show_port_pair = mock.Mock(
self.neutronclient.show_sfc_port_pair = mock.Mock(
return_value=self._port_pair
)
@ -295,7 +295,7 @@ class TestShowSfcPortPair(fakes.TestNeutronClientOSCV2):
def test_show_port_pair(self):
client = self.app.client_manager.neutronclient
mock_port_pair_show = client.show_port_pair
mock_port_pair_show = client.show_sfc_port_pair
arglist = [
self._port_pair_id,
]

36
neutronclient/tests/unit/osc/v2/sfc/test_port_pair_group.py
View File

@ -51,7 +51,7 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.create_port_pair_group = mock.Mock(
self.neutronclient.create_sfc_port_pair_group = mock.Mock(
return_value={'port_pair_group': self._port_pair_group})
self.data = self.get_data()
# Get the command object to test
@ -70,7 +70,7 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_pair_group.assert_called_once_with({
self.neutronclient.create_sfc_port_pair_group.assert_called_once_with({
'port_pair_group': {
'name': self._port_pair_group['name'],
'port_pairs': [self._port_pair_group['port_pairs']]}
@ -92,7 +92,7 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns, data = (self.cmd.take_action(parsed_args))
self.neutronclient.create_port_pair_group.assert_called_once_with({
self.neutronclient.create_sfc_port_pair_group.assert_called_once_with({
'port_pair_group': {
'name': self._port_pair_group['name'],
'port_pairs': [self._port_pair_group['port_pairs']],
@ -113,14 +113,14 @@ class TestDeleteSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.delete_port_pair_group = mock.Mock(
self.neutronclient.delete_sfc_port_pair_group = mock.Mock(
return_value=None)
self.cmd = sfc_port_pair_group.DeleteSfcPortPairGroup(self.app,
self.namespace)
def test_delete_port_pair_group(self):
client = self.app.client_manager.neutronclient
mock_port_pair_group_delete = client.delete_port_pair_group
mock_port_pair_group_delete = client.delete_sfc_port_pair_group
arglist = [
self._port_pair_group[0]['id'],
]
@ -162,19 +162,20 @@ class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.list_port_pair_group = mock.Mock(
self.neutronclient.list_sfc_port_pair_groups = mock.Mock(
return_value={'port_pair_groups': self._ppgs}
)
# Get the command object to test
self.cmd = sfc_port_pair_group.ListSfcPortPairGroup(self.app,
self.namespace)
def test_list_port_pair_group(self):
def test_list_port_pair_groups(self):
arglist = []
verifylist = []
parsed_args = self.check_parser(self.cmd, arglist, verifylist)
columns = self.cmd.take_action(parsed_args)[0]
ppgs = self.neutronclient.list_port_pair_group()['port_pair_groups']
ppgs = self.neutronclient \
.list_sfc_port_pair_groups()['port_pair_groups']
ppg = ppgs[0]
data = [
ppg['id'],
@ -188,7 +189,8 @@ class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
def test_list_with_long_option(self):
arglist = ['--long']
verifylist = [('long', True)]
ppgs = self.neutronclient.list_port_pair_group()['port_pair_groups']
ppgs = self.neutronclient \
.list_sfc_port_pair_groups()['port_pair_groups']
ppg = ppgs[0]
data = [
ppg['id'],
@ -217,9 +219,9 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.update_port_pair_group = mock.Mock(
self.neutronclient.update_sfc_port_pair_group = mock.Mock(
return_value=None)
self.mocked = self.neutronclient.update_port_pair_group
self.mocked = self.neutronclient.update_sfc_port_pair_group
self.cmd = sfc_port_pair_group.SetSfcPortPairGroup(self.app,
self.namespace)
@ -265,7 +267,7 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
def test_set_no_port_pair(self):
client = self.app.client_manager.neutronclient
mock_port_pair_group_update = client.update_port_pair_group
mock_port_pair_group_update = client.update_sfc_port_pair_group
arglist = [
self._port_pair_group_name,
'--name', 'name_updated',
@ -317,7 +319,7 @@ class TestShowSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.show_port_pair_group = mock.Mock(
self.neutronclient.show_sfc_port_pair_group = mock.Mock(
return_value=self._port_pair_group
)
self.cmd = sfc_port_pair_group.ShowSfcPortPairGroup(self.app,
@ -325,7 +327,7 @@ class TestShowSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
def test_show_port_pair_group(self):
client = self.app.client_manager.neutronclient
mock_port_pair_group_show = client.show_port_pair_group
mock_port_pair_group_show = client.show_sfc_port_pair_group
arglist = [
self._port_pair_group_id,
]
@ -354,9 +356,9 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
mock.patch(
'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id',
new=_get_id).start()
self.neutronclient.update_port_pair_group = mock.Mock(
self.neutronclient.update_sfc_port_pair_group = mock.Mock(
return_value=None)
self.mocked = self.neutronclient.update_port_pair_group
self.mocked = self.neutronclient.update_sfc_port_pair_group
self.cmd = sfc_port_pair_group.UnsetSfcPortPairGroup(
self.app, self.namespace)
@ -406,7 +408,7 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2):
def test_unset_all_port_pair(self):
client = self.app.client_manager.neutronclient
mock_port_pair_group_update = client.update_port_pair_group
mock_port_pair_group_update = client.update_sfc_port_pair_group
arglist = [
self._port_pair_group_name,
'--all-port-pair',

40
neutronclient/v2_0/client.py
View File

@ -2171,91 +2171,91 @@ class Client(ClientBase):
return self.delete(
self.bgpvpn_router_association_path % (bgpvpn, router_assoc))
def create_port_pair(self, body=None):
def create_sfc_port_pair(self, body=None):
"""Creates a new Port Pair."""
return self.post(self.sfc_port_pairs_path, body=body)
def update_port_pair(self, port_pair, body=None):
def update_sfc_port_pair(self, port_pair, body=None):
"""Update a Port Pair."""
return self.put(self.sfc_port_pair_path % port_pair, body=body)
def delete_port_pair(self, port_pair):
def delete_sfc_port_pair(self, port_pair):
"""Deletes the specified Port Pair."""
return self.delete(self.sfc_port_pair_path % (port_pair))
def list_port_pair(self, retrieve_all=True, **_params):
def list_sfc_port_pairs(self, retrieve_all=True, **_params):
"""Fetches a list of all Port Pairs."""
return self.list('port_pairs', self.sfc_port_pairs_path, retrieve_all,
**_params)
def show_port_pair(self, port_pair, **_params):
def show_sfc_port_pair(self, port_pair, **_params):
"""Fetches information of a certain Port Pair."""
return self.get(self.sfc_port_pair_path % (port_pair), params=_params)
def create_port_pair_group(self, body=None):
def create_sfc_port_pair_group(self, body=None):
"""Creates a new Port Pair Group."""
return self.post(self.sfc_port_pair_groups_path, body=body)
def update_port_pair_group(self, port_pair_group, body=None):
def update_sfc_port_pair_group(self, port_pair_group, body=None):
"""Update a Port Pair Group."""
return self.put(self.sfc_port_pair_group_path % port_pair_group,
body=body)
def delete_port_pair_group(self, port_pair_group):
def delete_sfc_port_pair_group(self, port_pair_group):
"""Deletes the specified Port Pair Group."""
return self.delete(self.sfc_port_pair_group_path % (port_pair_group))
def list_port_pair_group(self, retrieve_all=True, **_params):
def list_sfc_port_pair_groups(self, retrieve_all=True, **_params):
"""Fetches a list of all Port Pair Groups."""
return self.list('port_pair_groups', self.sfc_port_pair_groups_path,
retrieve_all, **_params)
def show_port_pair_group(self, port_pair_group, **_params):
def show_sfc_port_pair_group(self, port_pair_group, **_params):
"""Fetches information of a certain Port Pair Group."""
return self.get(self.sfc_port_pair_group_path % (port_pair_group),
params=_params)
def create_port_chain(self, body=None):
def create_sfc_port_chain(self, body=None):
"""Creates a new Port Chain."""
return self.post(self.sfc_port_chains_path, body=body)
def update_port_chain(self, port_chain, body=None):
def update_sfc_port_chain(self, port_chain, body=None):
"""Update a Port Chain."""
return self.put(self.sfc_port_chain_path % port_chain, body=body)
def delete_port_chain(self, port_chain):
def delete_sfc_port_chain(self, port_chain):
"""Deletes the specified Port Chain."""
return self.delete(self.sfc_port_chain_path % (port_chain))
def list_port_chain(self, retrieve_all=True, **_params):
def list_sfc_port_chains(self, retrieve_all=True, **_params):
"""Fetches a list of all Port Chains."""
return self.list('port_chains', self.sfc_port_chains_path,
retrieve_all, **_params)
def show_port_chain(self, port_chain, **_params):
def show_sfc_port_chain(self, port_chain, **_params):
"""Fetches information of a certain Port Chain."""
return self.get(self.sfc_port_chain_path % (port_chain),
params=_params)
def create_flow_classifier(self, body=None):
def create_sfc_flow_classifier(self, body=None):
"""Creates a new Flow Classifier."""
return self.post(self.sfc_flow_classifiers_path, body=body)
def update_flow_classifier(self, flow_classifier, body=None):
def update_sfc_flow_classifier(self, flow_classifier, body=None):
"""Update a Flow Classifier."""
return self.put(self.sfc_flow_classifier_path % flow_classifier,
body=body)
def delete_flow_classifier(self, flow_classifier):
def delete_sfc_flow_classifier(self, flow_classifier):
"""Deletes the specified Flow Classifier."""
return self.delete(self.sfc_flow_classifier_path % (flow_classifier))
def list_flow_classifier(self, retrieve_all=True, **_params):
def list_sfc_flow_classifiers(self, retrieve_all=True, **_params):
"""Fetches a list of all Flow Classifiers."""
return self.list('flow_classifiers', self.sfc_flow_classifiers_path,
retrieve_all, **_params)
def show_flow_classifier(self, flow_classifier, **_params):
def show_sfc_flow_classifier(self, flow_classifier, **_params):
"""Fetches information of a certain Flow Classifier."""
return self.get(self.sfc_flow_classifier_path % (flow_classifier),
params=_params)