diff --git a/neutronclient/osc/v2/sfc/sfc_flow_classifier.py b/neutronclient/osc/v2/sfc/sfc_flow_classifier.py index d223674c4..775730fc2 100755 --- a/neutronclient/osc/v2/sfc/sfc_flow_classifier.py +++ b/neutronclient/osc/v2/sfc/sfc_flow_classifier.py @@ -55,6 +55,26 @@ _attr_map = ( ('project_id', 'Project', column_util.LIST_LONG_ONLY), ) +_attr_map_dict = { + 'id': 'ID', + 'name': 'Name', + 'description': 'Description', + 'summary': 'Summary', + 'protocol': 'Protocol', + 'ethertype': 'Ethertype', + 'source_ip_prefix': 'Source IP', + 'destination_ip_prefix': 'Destination IP', + 'logical_source_port': 'Logical Source Port', + 'logical_destination_port': 'Logical Destination Port', + 'source_port_range_min': 'Source Port Range Min', + 'source_port_range_max': 'Source Port Range Max', + 'destination_port_range_min': 'Destination Port Range Min', + 'destination_port_range_max': 'Destination Port Range Max', + 'l7_parameters': 'L7 Parameters', + 'tenant_id': 'Project', + 'project_id': 'Project', +} + class CreateSfcFlowClassifier(command.ShowOne): _description = _("Create a flow classifier") @@ -114,11 +134,11 @@ class CreateSfcFlowClassifier(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network attrs = _get_common_attrs(self.app.client_manager, parsed_args) - body = {resource: attrs} - obj = client.create_sfc_flow_classifier(body)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + obj = client.create_sfc_flow_classifier(**attrs) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id', 'summary']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -131,20 +151,27 @@ class DeleteSfcFlowClassifier(command.Command): parser.add_argument( 'flow_classifier', metavar='', - help=_("Flow classifier to delete (name or ID)") + nargs='+', + help=_("Flow classifier(s) to delete (name or ID)") ) return parser def take_action(self, parsed_args): - # TODO(mohan): Add support for deleting multiple resources. - client = self.app.client_manager.neutronclient - fc_id = _get_id(client, parsed_args.flow_classifier, resource) - try: - 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") - % {'fc': parsed_args.flow_classifier, 'e': e}) + client = self.app.client_manager.network + result = 0 + for fcl in parsed_args.flow_classifier: + try: + fc_id = client.find_sfc_flow_classifier( + fcl, ignore_missing=False)['id'] + client.delete_sfc_flow_classifier(fc_id) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete flow classifier with name " + "or ID '%(fc)s': %(e)s"), {'fc': fcl, 'e': e}) + if result > 0: + total = len(parsed_args.flow_classifier) + msg = (_("%(result)s of %(total)s flow classifier(s) " + "failed to delete.") % {'result': result, 'total': total}) raise exceptions.CommandError(msg) @@ -161,8 +188,8 @@ class ListSfcFlowClassifier(command.Lister): return parser def extend_list(self, data, parsed_args): - ext_data = data['flow_classifiers'] - for d in ext_data: + ext_data = [] + for d in data: val = [] protocol = d['protocol'].upper() if d['protocol'] else 'any' val.append('protocol: ' + protocol) @@ -180,6 +207,7 @@ class ListSfcFlowClassifier(command.Lister): l7_param = 'l7_parameters: {%s}' % ','.join(d['l7_parameters']) val.append(l7_param) d['summary'] = ',\n'.join(val) + ext_data.append(d) return ext_data def _get_protocol_port_details(self, data, val): @@ -197,8 +225,8 @@ class ListSfcFlowClassifier(command.Lister): val, ip_prefix, min_port, max_port) def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - obj = client.list_sfc_flow_classifiers() + client = self.app.client_manager.network + obj = client.sfc_flow_classifiers() obj_extend = self.extend_list(obj, parsed_args) headers, columns = column_util.get_column_definitions( _attr_map, long_listing=parsed_args.long) @@ -227,13 +255,13 @@ class SetSfcFlowClassifier(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - fc_id = _get_id(client, parsed_args.flow_classifier, resource) + client = self.app.client_manager.network + fc_id = client.find_sfc_flow_classifier(parsed_args.flow_classifier, + ignore_missing=False)['id'] attrs = _get_common_attrs(self.app.client_manager, parsed_args, is_create=False) - body = {resource: attrs} try: - client.update_sfc_flow_classifier(fc_id, body) + client.update_sfc_flow_classifier(fc_id, **attrs) except Exception as e: msg = (_("Failed to update flow classifier '%(fc)s': %(e)s") % {'fc': parsed_args.flow_classifier, 'e': e}) @@ -253,10 +281,12 @@ class ShowSfcFlowClassifier(command.ShowOne): return parser 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_sfc_flow_classifier(fc_id)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + client = self.app.client_manager.network + fc_id = client.find_sfc_flow_classifier(parsed_args.flow_classifier, + ignore_missing=False)['id'] + obj = client.get_sfc_flow_classifier(fc_id) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id', 'summary']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -282,13 +312,13 @@ def _get_attrs(client_manager, attrs, parsed_args): if parsed_args.destination_ip_prefix is not None: attrs['destination_ip_prefix'] = parsed_args.destination_ip_prefix if parsed_args.logical_source_port is not None: - attrs['logical_source_port'] = _get_id( - client_manager.neutronclient, parsed_args.logical_source_port, - 'port') + attrs['logical_source_port'] = client_manager.network.find_port( + parsed_args.logical_source_port, ignore_missing=False + )['id'] if parsed_args.logical_destination_port is not None: - attrs['logical_destination_port'] = _get_id( - client_manager.neutronclient, parsed_args.logical_destination_port, - 'port') + attrs['logical_destination_port'] = client_manager.network.find_port( + parsed_args.logical_destination_port, ignore_missing=False + )['id'] if parsed_args.source_port is not None: _fill_protocol_port_info(attrs, 'source', parsed_args.source_port) @@ -314,7 +344,3 @@ def _fill_protocol_port_info(attrs, port_type, port_val): message = (_("Protocol port value %s must be an integer " "or integer:integer.") % port_val) raise nc_exc.CommandError(message=message) - - -def _get_id(client, id_or_name, resource): - return client.find_resource(resource, id_or_name)['id'] diff --git a/neutronclient/osc/v2/sfc/sfc_port_chain.py b/neutronclient/osc/v2/sfc/sfc_port_chain.py index 4ce790dc3..d96007b85 100755 --- a/neutronclient/osc/v2/sfc/sfc_port_chain.py +++ b/neutronclient/osc/v2/sfc/sfc_port_chain.py @@ -36,10 +36,20 @@ _attr_map = ( ('chain_parameters', 'Chain Parameters', column_util.LIST_BOTH), ('description', 'Description', column_util.LIST_LONG_ONLY), - ('chain_id', 'Chain ID', column_util.LIST_BOTH), ('project_id', 'Project', column_util.LIST_LONG_ONLY), ) +_attr_map_dict = { + 'id': 'ID', + 'name': 'Name', + 'port_pair_groups': 'Port Pair Groups', + 'flow_classifiers': 'Flow Classifiers', + 'chain_parameters': 'Chain Parameters', + 'description': 'Description', + 'tenant_id': 'Project', + 'project_id': 'Project', +} + class CreateSfcPortChain(command.ShowOne): _description = _("Create a port chain") @@ -81,11 +91,11 @@ class CreateSfcPortChain(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network attrs = _get_common_attrs(self.app.client_manager, parsed_args) - body = {resource: attrs} - obj = client.create_sfc_port_chain(body)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + obj = client.create_sfc_port_chain(**attrs) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -98,20 +108,28 @@ class DeleteSfcPortChain(command.Command): parser.add_argument( 'port_chain', metavar="", - help=_("Port chain to delete (name or ID)") + nargs='+', + help=_("Port chain(s) to delete (name or ID)") ) return parser def take_action(self, parsed_args): - # TODO(mohan): Add support for deleting multiple resources. - client = self.app.client_manager.neutronclient - pc_id = _get_id(client, parsed_args.port_chain, resource) - try: - 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") - % {'pc': parsed_args.port_chain, 'e': e}) + client = self.app.client_manager.network + result = 0 + for pc in parsed_args.port_chain: + try: + pc_id = client.find_sfc_port_chain( + pc, ignore_missing=False)['id'] + client.delete_sfc_port_chain(pc_id) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete port chain with name " + "or ID '%(pc)s': %(e)s"), {'pc': pc, 'e': e}) + if result > 0: + total = len(parsed_args.port_chain) + msg = (_("%(result)s of %(total)s port chain(s) " + "failed to delete.") % {'result': result, + 'total': total}) raise exceptions.CommandError(msg) @@ -129,13 +147,13 @@ class ListSfcPortChain(command.Lister): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - data = client.list_sfc_port_chains() + client = self.app.client_manager.network + data = client.sfc_port_chains() headers, columns = column_util.get_column_definitions( _attr_map, long_listing=parsed_args.long) return (headers, (utils.get_dict_properties(s, columns) - for s in data['port_chains'])) + for s in data)) class SetSfcPortChain(command.Command): @@ -184,8 +202,9 @@ class SetSfcPortChain(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - pc_id = _get_id(client, parsed_args.port_chain, resource) + client = self.app.client_manager.network + pc_id = client.find_sfc_port_chain(parsed_args.port_chain, + ignore_missing=False)['id'] attrs = _get_common_attrs(self.app.client_manager, parsed_args, is_create=False) if parsed_args.no_flow_classifier: @@ -194,13 +213,14 @@ class SetSfcPortChain(command.Command): if parsed_args.no_flow_classifier: fc_list = [] else: - fc_list = client.find_resource( - resource, parsed_args.port_chain, - cmd_resource='sfc_port_chain')['flow_classifiers'] + fc_list = client.find_sfc_port_chain( + parsed_args.port_chain, + ignore_missing=False + )['flow_classifiers'] for fc in parsed_args.flow_classifiers: - fc_id = client.find_resource( - 'flow_classifier', fc, - cmd_resource='sfc_flow_classifier')['id'] + fc_id = client.find_sfc_flow_classifier( + fc, + ignore_missing=False)['id'] if fc_id not in fc_list: fc_list.append(fc_id) attrs['flow_classifiers'] = fc_list @@ -211,27 +231,25 @@ class SetSfcPortChain(command.Command): if parsed_args.no_port_pair_group and parsed_args.port_pair_groups: ppg_list = [] for ppg in parsed_args.port_pair_groups: - ppg_id = client.find_resource( - 'port_pair_group', ppg, - cmd_resource='sfc_port_pair_group')['id'] + ppg_id = client.find_sfc_port_pair_group( + ppg, ignore_missing=False)['id'] if ppg_id not in ppg_list: ppg_list.append(ppg_id) attrs['port_pair_groups'] = 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'] + ppg_list = client.find_sfc_port_chain( + parsed_args.port_chain, + ignore_missing=False + )['port_pair_groups'] for ppg in parsed_args.port_pair_groups: - ppg_id = client.find_resource( - 'port_pair_group', ppg, - cmd_resource='sfc_port_pair_group')['id'] + ppg_id = client.find_sfc_port_pair_group( + ppg, ignore_missing=False)['id'] if ppg_id not in ppg_list: ppg_list.append(ppg_id) attrs['port_pair_groups'] = ppg_list - body = {resource: attrs} try: - client.update_sfc_port_chain(pc_id, body) + client.update_sfc_port_chain(pc_id, **attrs) except Exception as e: msg = (_("Failed to update port chain '%(pc)s': %(e)s") % {'pc': parsed_args.port_chain, 'e': e}) @@ -251,10 +269,12 @@ class ShowSfcPortChain(command.ShowOne): return parser 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_sfc_port_chain(pc_id)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + client = self.app.client_manager.network + pc_id = client.find_sfc_port_chain(parsed_args.port_chain, + ignore_missing=False)['id'] + obj = client.get_sfc_port_chain(pc_id) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -290,30 +310,31 @@ class UnsetSfcPortChain(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - pc_id = _get_id(client, parsed_args.port_chain, resource) + client = self.app.client_manager.network + pc_id = client.find_sfc_port_chain(parsed_args.port_chain, + ignore_missing=False)['id'] attrs = {} if parsed_args.flow_classifiers: - fc_list = client.find_resource( - resource, parsed_args.port_chain, - cmd_resource='sfc_port_chain')['flow_classifiers'] + fc_list = client.find_sfc_port_chain( + parsed_args.port_chain, ignore_missing=False + )['flow_classifiers'] for fc in parsed_args.flow_classifiers: - fc_id = client.find_resource( - 'flow_classifier', fc, - cmd_resource='sfc_flow_classifier')['id'] + fc_id = client.find_sfc_flow_classifier( + fc, + ignore_missing=False)['id'] if fc_id in fc_list: fc_list.remove(fc_id) attrs['flow_classifiers'] = fc_list if parsed_args.all_flow_classifier: attrs['flow_classifiers'] = [] if parsed_args.port_pair_groups: - ppg_list = client.find_resource( - resource, parsed_args.port_chain, - cmd_resource='sfc_port_chain')['port_pair_groups'] + ppg_list = client.find_sfc_port_chain( + parsed_args.port_chain, + ignore_missing=False)['port_pair_groups'] for ppg in parsed_args.port_pair_groups: - ppg_id = client.find_resource( - 'port_pair_group', ppg, - cmd_resource='sfc_port_pair_group')['id'] + ppg_id = client.find_sfc_port_pair_group( + ppg, + ignore_missing=False)['id'] if ppg_id in ppg_list: ppg_list.remove(ppg_id) if ppg_list == []: @@ -321,9 +342,8 @@ class UnsetSfcPortChain(command.Command): ' specified.') raise exceptions.CommandError(message) attrs['port_pair_groups'] = ppg_list - body = {resource: attrs} try: - client.update_sfc_port_chain(pc_id, body) + client.update_sfc_port_chain(pc_id, **attrs) except Exception as e: msg = (_("Failed to unset port chain '%(pc)s': %(e)s") % {'pc': parsed_args.port_chain, 'e': e}) @@ -332,17 +352,18 @@ class UnsetSfcPortChain(command.Command): def _get_common_attrs(client_manager, parsed_args, is_create=True): attrs = {} + client = client_manager.network if parsed_args.name is not None: attrs['name'] = parsed_args.name if parsed_args.description is not None: attrs['description'] = parsed_args.description if parsed_args.port_pair_groups: - attrs['port_pair_groups'] = [(_get_id(client_manager.neutronclient, - ppg, 'port_pair_group')) + attrs['port_pair_groups'] = [client.find_sfc_port_pair_group( + ppg, ignore_missing=False)['id'] for ppg in parsed_args.port_pair_groups] if parsed_args.flow_classifiers: - attrs['flow_classifiers'] = [(_get_id(client_manager.neutronclient, fc, - 'flow_classifier')) + attrs['flow_classifiers'] = [client.find_sfc_flow_classifier( + fc, ignore_missing=False)['id'] for fc in parsed_args.flow_classifiers] if is_create is True: _get_attrs(attrs, parsed_args) @@ -358,7 +379,3 @@ def _get_attrs(attrs, parsed_args): if 'symmetric' in chain_param: chain_params['symmetric'] = chain_param['symmetric'] attrs['chain_parameters'] = chain_params - - -def _get_id(client, id_or_name, resource): - return client.find_resource(resource, id_or_name)['id'] diff --git a/neutronclient/osc/v2/sfc/sfc_port_pair.py b/neutronclient/osc/v2/sfc/sfc_port_pair.py index 68d131cdd..11e8e7631 100755 --- a/neutronclient/osc/v2/sfc/sfc_port_pair.py +++ b/neutronclient/osc/v2/sfc/sfc_port_pair.py @@ -38,6 +38,17 @@ _attr_map = ( ('project_id', 'Project', column_util.LIST_LONG_ONLY), ) +_attr_map_dict = { + 'id': 'ID', + 'name': 'Name', + 'description': 'Description', + 'ingress': 'Ingress Logical Port', + 'egress': 'Egress Logical Port', + 'service_function_parameters': 'Service Function Parameters', + 'tenant_id': 'Project', + 'project_id': 'Project', +} + class CreateSfcPortPair(command.ShowOne): _description = _("Create a port pair") @@ -76,11 +87,11 @@ class CreateSfcPortPair(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network attrs = _get_common_attrs(self.app.client_manager, parsed_args) - body = {resource: attrs} - obj = client.create_sfc_port_pair(body)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + obj = client.create_sfc_port_pair(**attrs) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -93,20 +104,29 @@ class DeleteSfcPortPair(command.Command): parser.add_argument( 'port_pair', metavar="", - help=_("Port pair to delete (name or ID)") + nargs='+', + help=_("Port pair(s) to delete (name or ID)") ) return parser def take_action(self, parsed_args): - # TODO(mohan): Add support for deleting multiple resources. - client = self.app.client_manager.neutronclient - port_pair_id = _get_id(client, parsed_args.port_pair, resource) - try: - 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") - % {'port_pair': parsed_args.port_pair, 'e': e}) + client = self.app.client_manager.network + result = 0 + for pp in parsed_args.port_pair: + try: + port_pair_id = client.find_sfc_port_pair( + pp, ignore_missing=False)['id'] + client.delete_sfc_port_pair(port_pair_id) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete port pair with name " + "or ID '%(port_pair)s': %(e)s"), + {'port_pair': pp, 'e': e}) + if result > 0: + total = len(parsed_args.port_pair) + msg = (_("%(result)s of %(total)s port pair(s) " + "failed to delete.") % {'result': result, + 'total': total}) raise exceptions.CommandError(msg) @@ -123,14 +143,14 @@ class ListSfcPortPair(command.Lister): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - data = client.list_sfc_port_pairs() + client = self.app.client_manager.network + data = client.sfc_port_pairs() headers, columns = column_util.get_column_definitions( _attr_map, long_listing=parsed_args.long) return (headers, (utils.get_dict_properties( s, columns, - ) for s in data['port_pairs'])) + ) for s in data)) class SetSfcPortPair(command.Command): @@ -154,13 +174,14 @@ class SetSfcPortPair(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - port_pair_id = _get_id(client, parsed_args.port_pair, resource) + client = self.app.client_manager.network + port_pair_id = client.find_sfc_port_pair( + parsed_args.port_pair, ignore_missing=False + )['id'] attrs = _get_common_attrs(self.app.client_manager, parsed_args, is_create=False) - body = {resource: attrs} try: - client.update_sfc_port_pair(port_pair_id, body) + client.update_sfc_port_pair(port_pair_id, **attrs) except Exception as e: msg = (_("Failed to update port pair '%(port_pair)s': %(e)s") % {'port_pair': parsed_args.port_pair, 'e': e}) @@ -180,10 +201,13 @@ class ShowSfcPortPair(command.ShowOne): return parser 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_sfc_port_pair(port_pair_id)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + client = self.app.client_manager.network + port_pair_id = client.find_sfc_port_pair( + parsed_args.port_pair, ignore_missing=False + )['id'] + obj = client.get_sfc_port_pair(port_pair_id) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -200,12 +224,15 @@ def _get_common_attrs(client_manager, parsed_args, is_create=True): def _get_attrs(client_manager, attrs, parsed_args): + client = client_manager.network if parsed_args.ingress is not None: - attrs['ingress'] = _get_id(client_manager.neutronclient, - parsed_args.ingress, 'port') + attrs['ingress'] = client.find_port( + parsed_args.ingress, ignore_missing=False + )['id'] if parsed_args.egress is not None: - attrs['egress'] = _get_id(client_manager.neutronclient, - parsed_args.egress, 'port') + attrs['egress'] = client.find_port( + parsed_args.egress, ignore_missing=False + )['id'] if parsed_args.service_function_parameters is not None: attrs['service_function_parameters'] = _get_service_function_params( parsed_args.service_function_parameters) @@ -222,7 +249,3 @@ def _get_service_function_params(sf_params): if 'weight' in sf_param: attrs['weight'] = sf_param['weight'] return attrs - - -def _get_id(client, id_or_name, resource): - return client.find_resource(resource, id_or_name)['id'] diff --git a/neutronclient/osc/v2/sfc/sfc_port_pair_group.py b/neutronclient/osc/v2/sfc/sfc_port_pair_group.py index 8e7fadd26..5ad3ebea7 100755 --- a/neutronclient/osc/v2/sfc/sfc_port_pair_group.py +++ b/neutronclient/osc/v2/sfc/sfc_port_pair_group.py @@ -34,11 +34,21 @@ _attr_map = ( ('port_pair_group_parameters', 'Port Pair Group Parameters', column_util.LIST_BOTH), ('description', 'Description', column_util.LIST_LONG_ONLY), - ('group_id', 'Loadbalance ID', column_util.LIST_LONG_ONLY), ('project_id', 'Project', column_util.LIST_LONG_ONLY), - ('tap_enabled', 'Tap Enabled', column_util.LIST_BOTH) + ('is_tap_enabled', 'Tap Enabled', column_util.LIST_BOTH) ) +_attr_map_dict = { + 'id': 'ID', + 'name': 'Name', + 'description': 'Description', + 'port_pairs': 'Port Pair', + 'port_pair_group_parameters': 'Port Pair Group Parameters', + 'is_tap_enabled': 'Tap Enabled', + 'tenant_id': 'Project', + 'project_id': 'Project', +} + class CreateSfcPortPairGroup(command.ShowOne): _description = _("Create a port pair group") @@ -85,11 +95,11 @@ class CreateSfcPortPairGroup(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network attrs = _get_common_attrs(self.app.client_manager, parsed_args) - body = {resource: attrs} - obj = client.create_sfc_port_pair_group(body)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + obj = client.create_sfc_port_pair_group(**attrs) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -102,20 +112,28 @@ class DeleteSfcPortPairGroup(command.Command): parser.add_argument( 'port_pair_group', metavar='', - help=_("Port pair group to delete (name or ID)") + nargs='+', + help=_("Port pair group(s) to delete (name or ID)") ) return parser def take_action(self, parsed_args): - # TODO(mohan): Add support for deleting multiple resources. - client = self.app.client_manager.neutronclient - ppg_id = _get_id(client, parsed_args.port_pair_group, resource) - try: - 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") - % {'ppg': parsed_args.port_pair_group, 'e': e}) + client = self.app.client_manager.network + result = 0 + for ppg in parsed_args.port_pair_group: + try: + ppg_id = client.find_sfc_port_pair_group( + ppg, ignore_missing=False)['id'] + client.delete_sfc_port_pair_group(ppg_id) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete port pair group with name " + "or ID '%(ppg)s': %(e)s"), {'ppg': ppg, 'e': e}) + if result > 0: + total = len(parsed_args.port_pair_group) + msg = (_("%(result)s of %(total)s port pair group(s) " + "failed to delete.") % {'result': result, + 'total': total}) raise exceptions.CommandError(msg) @@ -133,14 +151,14 @@ class ListSfcPortPairGroup(command.Lister): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - data = client.list_sfc_port_pair_groups() + client = self.app.client_manager.network + data = client.sfc_port_pair_groups() headers, columns = column_util.get_column_definitions( _attr_map, long_listing=parsed_args.long) return (headers, (utils.get_dict_properties( s, columns, - ) for s in data['port_pair_groups'])) + ) for s in data)) class SetSfcPortPairGroup(command.Command): @@ -175,26 +193,26 @@ class SetSfcPortPairGroup(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - ppg_id = _get_id(client, parsed_args.port_pair_group, resource) + client = self.app.client_manager.network + ppg_id = client.find_sfc_port_pair_group( + parsed_args.port_pair_group)['id'] attrs = _get_common_attrs(self.app.client_manager, parsed_args, is_create=False) if parsed_args.no_port_pair: attrs['port_pairs'] = [] if parsed_args.port_pairs: - added = [client.find_resource('port_pair', pp, - cmd_resource='sfc_port_pair')['id'] + added = [client.find_sfc_port_pair(pp, + ignore_missing=False)['id'] for pp in parsed_args.port_pairs] if parsed_args.no_port_pair: existing = [] else: - existing = client.find_resource( - resource, parsed_args.port_pair_group, - cmd_resource='sfc_port_pair_group')['port_pairs'] + existing = client.find_sfc_port_pair_group( + parsed_args.port_pair_group, + ignore_missing=False)['port_pairs'] attrs['port_pairs'] = sorted(list(set(existing) | set(added))) - body = {resource: attrs} try: - client.update_sfc_port_pair_group(ppg_id, body) + client.update_sfc_port_pair_group(ppg_id, **attrs) except Exception as e: msg = (_("Failed to update port pair group '%(ppg)s': %(e)s") % {'ppg': parsed_args.port_pair_group, 'e': e}) @@ -214,10 +232,12 @@ class ShowSfcPortPairGroup(command.ShowOne): return parser 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_sfc_port_pair_group(ppg_id)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + client = self.app.client_manager.network + ppg_id = client.find_sfc_port_pair_group( + parsed_args.port_pair_group, ignore_missing=False)['id'] + obj = client.get_sfc_port_pair_group(ppg_id) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -246,22 +266,22 @@ class UnsetSfcPortPairGroup(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - ppg_id = _get_id(client, parsed_args.port_pair_group, resource) + client = self.app.client_manager.network + ppg_id = client.find_sfc_port_pair_group( + parsed_args.port_pair_group, ignore_missing=False)['id'] attrs = {} if parsed_args.port_pairs: - existing = client.find_resource( - resource, parsed_args.port_pair_group, - cmd_resource='sfc_port_pair_group')['port_pairs'] - removed = [client.find_resource('port_pair', pp, - cmd_resource='sfc_port_pair')['id'] + existing = client.find_sfc_port_pair_group( + parsed_args.port_pair_group, + ignore_missing=False)['port_pairs'] + removed = [client.find_sfc_port_pair(pp, + ignore_missing=False)['id'] for pp in parsed_args.port_pairs] attrs['port_pairs'] = list(set(existing) - set(removed)) if parsed_args.all_port_pair: attrs['port_pairs'] = [] - body = {resource: attrs} try: - client.update_sfc_port_pair_group(ppg_id, body) + client.update_sfc_port_pair_group(ppg_id, **attrs) except Exception as e: msg = (_("Failed to unset port pair group '%(ppg)s': %(e)s") % {'ppg': parsed_args.port_pair_group, 'e': e}) @@ -280,15 +300,17 @@ def _get_ppg_param(attrs, ppg): def _get_common_attrs(client_manager, parsed_args, is_create=True): + client = client_manager.network attrs = {} if parsed_args.name is not None: attrs['name'] = parsed_args.name if parsed_args.description is not None: attrs['description'] = parsed_args.description if parsed_args.port_pairs: - attrs['port_pairs'] = [(_get_id(client_manager.neutronclient, pp, - 'port_pair')) + attrs['port_pairs'] = [client.find_sfc_port_pair( + pp, ignore_missing=False)['id'] for pp in parsed_args.port_pairs] + if is_create: _get_attrs(attrs, parsed_args) return attrs @@ -302,7 +324,3 @@ def _get_attrs(attrs, parsed_args): attrs['tap_enabled'] = True if parsed_args.disable_tap: attrs['tap_enabled'] = False - - -def _get_id(client, id_or_name, resource): - return client.find_resource(resource, id_or_name)['id'] diff --git a/neutronclient/osc/v2/sfc/sfc_service_graph.py b/neutronclient/osc/v2/sfc/sfc_service_graph.py index fc4772675..6c87d95fe 100644 --- a/neutronclient/osc/v2/sfc/sfc_service_graph.py +++ b/neutronclient/osc/v2/sfc/sfc_service_graph.py @@ -33,6 +33,15 @@ _attr_map = ( ('project_id', 'Project', column_util.LIST_LONG_ONLY), ) +_attr_map_dict = { + 'id': 'ID', + 'name': 'Name', + 'description': 'Description', + 'port_chains': 'Branching Points', + 'tenant_id': 'Project', + 'project_id': 'Project', +} + class CreateSfcServiceGraph(command.ShowOne): """Create a service graph.""" @@ -57,12 +66,13 @@ class CreateSfcServiceGraph(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network attrs = _get_common_attrs(self.app.client_manager, parsed_args) try: - body = {resource: attrs} - obj = client.create_sfc_service_graph(body)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + obj = client.create_sfc_service_graph(**attrs) + display_columns, columns = \ + utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data except Exception as e: @@ -92,13 +102,13 @@ class SetSfcServiceGraph(command.Command): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - service_graph_id = _get_id(client, parsed_args.service_graph, resource) + client = self.app.client_manager.network + service_graph_id = client.find_sfc_service_graph( + parsed_args.service_graph, ignore_missing=False)['id'] attrs = _get_common_attrs(self.app.client_manager, parsed_args, is_create=False) - body = {resource: attrs} try: - client.update_sfc_service_graph(service_graph_id, body) + client.update_sfc_service_graph(service_graph_id, **attrs) except Exception as e: msg = (_("Failed to update service graph " "'%(service_graph)s': %(e)s") @@ -114,14 +124,30 @@ class DeleteSfcServiceGraph(command.Command): parser.add_argument( 'service_graph', metavar="", - help=_("ID or name of the service graph to delete.") + nargs='+', + help=_("ID or name of the service graph(s) to delete.") ) return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - id = _get_id(client, parsed_args.service_graph, resource) - client.delete_sfc_service_graph(id) + client = self.app.client_manager.network + result = 0 + for sg in parsed_args.service_graph: + try: + sg_id = client.find_sfc_service_graph( + sg, ignore_missing=False)['id'] + client.delete_sfc_service_graph(sg_id) + except Exception as e: + result += 1 + LOG.error(_("Failed to delete service graph with name " + "or ID '%(sg)s': %(e)s"), + {'sg': sg, 'e': e}) + if result > 0: + total = len(parsed_args.service_graph) + msg = (_("%(result)s of %(total)s service graph(s) " + "failed to delete.") % {'result': result, + 'total': total}) + raise exceptions.CommandError(msg) class ListSfcServiceGraph(command.Lister): @@ -138,13 +164,13 @@ class ListSfcServiceGraph(command.Lister): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - data = client.list_sfc_service_graphs() + client = self.app.client_manager.network + data = client.sfc_service_graphs() headers, columns = column_util.get_column_definitions( _attr_map, long_listing=parsed_args.long) return (headers, (utils.get_dict_properties(s, columns) - for s in data['service_graphs'])) + for s in data)) class ShowSfcServiceGraph(command.ShowOne): @@ -160,10 +186,12 @@ class ShowSfcServiceGraph(command.ShowOne): return parser def take_action(self, parsed_args): - client = self.app.client_manager.neutronclient - sg_id = _get_id(client, parsed_args.service_graph, resource) - obj = client.show_sfc_service_graph(sg_id)[resource] - columns, display_columns = column_util.get_columns(obj, _attr_map) + client = self.app.client_manager.network + sg_id = client.find_sfc_service_graph(parsed_args.service_graph, + ignore_missing=False)['id'] + obj = client.get_sfc_service_graph(sg_id) + display_columns, columns = utils.get_osc_show_columns_for_sdk_resource( + obj, _attr_map_dict, ['location', 'tenant_id']) data = utils.get_dict_properties(obj, columns) return display_columns, data @@ -179,10 +207,10 @@ def _get_common_attrs(client_manager, parsed_args, is_create=True): return attrs -def _validate_destination_chains(comma_split, attrs, client_manager, sc_): +def _validate_destination_chains(comma_split, attrs, client, sc_): for e in comma_split: if e != "": - dc_ = _get_id(client_manager.neutronclient, e, 'port_chain') + dc_ = client.find_sfc_port_chain(e, ignore_missing=False)['id'] attrs['port_chains'][sc_].append(dc_) if _check_cycle(attrs['port_chains'], sc_, dc_): raise(exceptions.CommandError( @@ -214,6 +242,7 @@ def _visit(graph, src, new_dest, new_src): def _get_attrs_for_create(client_manager, attrs, parsed_args): + client = client_manager.network if parsed_args.branching_points: attrs['port_chains'] = {} src_chain = None @@ -224,8 +253,8 @@ def _get_attrs_for_create(client_manager, attrs, parsed_args): "destination chain for each source chain.") colon_split = c.split(':') src_chain = colon_split.pop(0) - sc_ = _get_id(client_manager.neutronclient, - src_chain, 'port_chain') + sc_ = client.find_sfc_port_chain(src_chain, + ignore_missing=False)['id'] for i in colon_split: comma_split = i.split(',') unique = set(comma_split) @@ -240,8 +269,4 @@ def _get_attrs_for_create(client_manager, attrs, parsed_args): "use already ".format(src_chain)) attrs['port_chains'][sc_] = [] _validate_destination_chains( - comma_split, attrs, client_manager, sc_) - - -def _get_id(client, id_or_name, resource): - return client.find_resource(resource, id_or_name)['id'] + comma_split, attrs, client, sc_) diff --git a/neutronclient/tests/unit/osc/v2/sfc/fakes.py b/neutronclient/tests/unit/osc/v2/sfc/fakes.py index ba07107fd..390c67e4e 100755 --- a/neutronclient/tests/unit/osc/v2/sfc/fakes.py +++ b/neutronclient/tests/unit/osc/v2/sfc/fakes.py @@ -14,13 +14,18 @@ # under the License. import argparse -import copy from unittest import mock from osc_lib.tests import utils from oslo_utils import uuidutils +from openstack.network.v2 import sfc_flow_classifier as flow_classifier +from openstack.network.v2 import sfc_port_chain as port_chain +from openstack.network.v2 import sfc_port_pair as port_pair +from openstack.network.v2 import sfc_port_pair_group as port_pair_group +from openstack.network.v2 import sfc_service_graph as service_graph + class TestNeutronClientOSCV2(utils.TestCommand): @@ -28,12 +33,32 @@ class TestNeutronClientOSCV2(utils.TestCommand): super(TestNeutronClientOSCV2, self).setUp() self.namespace = argparse.Namespace() self.app.client_manager.session = mock.Mock() - self.app.client_manager.neutronclient = mock.Mock() - self.neutronclient = self.app.client_manager.neutronclient - self.neutronclient.find_resource = mock.Mock( - side_effect=lambda resource, name_or_id, project_id=None, - cmd_resource=None, parent_id=None, fields=None: - {'id': name_or_id}) + self.app.client_manager.network = mock.Mock() + self.network = self.app.client_manager.network + self.network.find_sfc_flow_classifier = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_port_chain = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_port_pair = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_port_pair_group = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_service_graph = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_port = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) class FakeSfcPortPair(object): @@ -58,13 +83,14 @@ class FakeSfcPortPair(object): 'id': uuidutils.generate_uuid(), 'ingress': uuidutils.generate_uuid(), 'name': 'port-pair-name', - 'service_function_parameters': 'correlation=None,weight=1', + 'service_function_parameters': [('correlation', None), + ('weight', 1)], 'project_id': uuidutils.generate_uuid(), } # Overwrite default attributes. port_pair_attrs.update(attrs) - return copy.deepcopy(port_pair_attrs) + return port_pair.SfcPortPair(**port_pair_attrs) @staticmethod def create_port_pairs(attrs=None, count=1): @@ -102,18 +128,16 @@ class FakeSfcPortPairGroup(object): # Set default attributes. port_pair_group_attrs = { 'id': uuidutils.generate_uuid(), - 'group_id': uuidutils.generate_uuid(), 'name': 'port-pair-group-name', 'description': 'description', 'port_pairs': uuidutils.generate_uuid(), - 'port_pair_group_parameters': '{"lb_fields": []}', + 'port_pair_group_parameters': {"lb_fields": []}, 'project_id': uuidutils.generate_uuid(), 'tap_enabled': False } - # port_pair_group_attrs default attributes. port_pair_group_attrs.update(attrs) - return copy.deepcopy(port_pair_group_attrs) + return port_pair_group.SfcPortPairGroup(**port_pair_group_attrs) @staticmethod def create_port_pair_groups(attrs=None, count=1): @@ -164,10 +188,10 @@ class FakeSfcFlowClassifier(object): 'source_port_range_max': '20', 'source_port_range_min': '10', 'project_id': uuidutils.generate_uuid(), - 'l7_parameters': '{}' + 'l7_parameters': {} } flow_classifier_attrs.update(attrs) - return copy.deepcopy(flow_classifier_attrs) + return flow_classifier.SfcFlowClassifier(**flow_classifier_attrs) @staticmethod def create_flow_classifiers(attrs=None, count=1): @@ -205,18 +229,16 @@ class FakeSfcPortChain(object): # Set default attributes. port_chain_attrs = { 'id': uuidutils.generate_uuid(), - 'chain_id': uuidutils.generate_uuid(), 'name': 'port-chain-name', 'description': 'description', 'port_pair_groups': uuidutils.generate_uuid(), 'flow_classifiers': uuidutils.generate_uuid(), - 'chain_parameters': '{"correlation": mpls}', + 'chain_parameters': {"correlation": "mpls", "symmetric": False}, 'project_id': uuidutils.generate_uuid(), } - # port_pair_group_attrs default attributes. port_chain_attrs.update(attrs) - return copy.deepcopy(port_chain_attrs) + return port_chain.SfcPortChain(**port_chain_attrs) @staticmethod def create_port_chains(attrs=None, count=1): @@ -260,7 +282,7 @@ class FakeSfcServiceGraph(object): } service_graph_attrs.update(attrs) - return copy.deepcopy(service_graph_attrs) + return service_graph.SfcServiceGraph(**service_graph_attrs) @staticmethod def create_sfc_service_graphs(attrs=None, count=1): diff --git a/neutronclient/tests/unit/osc/v2/sfc/test_flow_classifier.py b/neutronclient/tests/unit/osc/v2/sfc/test_flow_classifier.py index 379f72767..6a9e691f9 100755 --- a/neutronclient/tests/unit/osc/v2/sfc/test_flow_classifier.py +++ b/neutronclient/tests/unit/osc/v2/sfc/test_flow_classifier.py @@ -15,15 +15,12 @@ from unittest import mock +from osc_lib import exceptions +import testtools + from neutronclient.osc.v2.sfc import sfc_flow_classifier from neutronclient.tests.unit.osc.v2.sfc import fakes -get_id = 'neutronclient.osc.v2.sfc.sfc_flow_classifier._get_id' - - -def _get_id(client, id_or_name, resource): - return id_or_name - class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): @@ -43,7 +40,8 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): 'Protocol', 'Source IP', 'Source Port Range Max', - 'Source Port Range Min') + 'Source Port Range Min', + 'Summary',) def get_data(self): return ( @@ -66,9 +64,8 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestCreateSfcFlowClassifier, self).setUp() - mock.patch(get_id, new=_get_id).start() - self.neutronclient.create_sfc_flow_classifier = mock.Mock( - return_value={'flow_classifier': self._fc}) + self.network.create_sfc_flow_classifier = mock.Mock( + return_value=self._fc) self.data = self.get_data() # Get the command object to test @@ -90,14 +87,13 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_flow_classifier.assert_called_once_with({ - 'flow_classifier': { - 'name': self._fc['name'], - 'logical_source_port': self._fc['logical_source_port'], - 'ethertype': self._fc['ethertype']} - }) + self.network.create_sfc_flow_classifier.assert_called_once_with( + **{'name': self._fc['name'], + 'logical_source_port': self._fc['logical_source_port'], + 'ethertype': self._fc['ethertype'] + } + ) self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) def test_create_flow_classifier(self): arglist = [ @@ -129,8 +125,8 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_flow_classifier.assert_called_once_with({ - 'flow_classifier': { + self.network.create_sfc_flow_classifier.assert_called_once_with( + **{ 'name': self._fc['name'], 'description': self._fc['description'], 'ethertype': self._fc['ethertype'], @@ -142,9 +138,8 @@ class TestCreateSfcFlowClassifier(fakes.TestNeutronClientOSCV2): self._fc['logical_destination_port'], 'l7_parameters': param } - }) + ) self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) class TestDeleteSfcFlowClassifier(fakes.TestNeutronClientOSCV2): @@ -154,20 +149,19 @@ class TestDeleteSfcFlowClassifier(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestDeleteSfcFlowClassifier, self).setUp() - mock.patch(get_id, new=_get_id).start() - self.neutronclient.delete_sfc_flow_classifier = mock.Mock( + self.network.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 + client = self.app.client_manager.network mock_flow_classifier_delete = client.delete_sfc_flow_classifier arglist = [ self._flow_classifier[0]['id'], ] verifylist = [ - ('flow_classifier', self._flow_classifier[0]['id']), + ('flow_classifier', [self._flow_classifier[0]['id']]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) @@ -175,6 +169,22 @@ class TestDeleteSfcFlowClassifier(fakes.TestNeutronClientOSCV2): self._flow_classifier[0]['id']) self.assertIsNone(result) + def test_delete_multiple_flow_classifiers_with_exception(self): + client = self.app.client_manager.network + target1 = self._flow_classifier[0]['id'] + arglist = [target1] + verifylist = [('flow_classifier', [target1])] + + client.find_sfc_flow_classifier.side_effect = [ + target1, exceptions.CommandError + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + msg = "1 of 2 flow classifier(s) failed to delete." + with testtools.ExpectedException(exceptions.CommandError) as e: + self.cmd.take_action(parsed_args) + self.assertEqual(msg, str(e)) + class TestSetSfcFlowClassifier(fakes.TestNeutronClientOSCV2): _flow_classifier = fakes.FakeSfcFlowClassifier.create_flow_classifier() @@ -183,14 +193,13 @@ class TestSetSfcFlowClassifier(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestSetSfcFlowClassifier, self).setUp() - mock.patch(get_id, new=_get_id).start() - self.neutronclient.update_sfc_flow_classifier = mock.Mock( + self.network.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 + client = self.app.client_manager.network mock_flow_classifier_update = client.update_sfc_flow_classifier arglist = [ self._flow_classifier_name, @@ -206,11 +215,11 @@ class TestSetSfcFlowClassifier(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'flow_classifier': { + attrs = { 'name': 'name_updated', - 'description': 'desc_updated'}} + 'description': 'desc_updated'} mock_flow_classifier_update.assert_called_once_with( - self._flow_classifier_name, attrs) + self._flow_classifier_name, **attrs) self.assertIsNone(result) @@ -234,7 +243,7 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2): _fc['source_port_range_max'], _fc['source_port_range_min'] ) - _flow_classifier = {'flow_classifier': _fc} + _flow_classifier = _fc _flow_classifier_id = _fc['id'] columns = ('Description', 'Destination IP', @@ -250,12 +259,12 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2): 'Protocol', 'Source IP', 'Source Port Range Max', - 'Source Port Range Min') + 'Source Port Range Min', + 'Summary',) def setUp(self): super(TestShowSfcFlowClassifier, self).setUp() - mock.patch(get_id, new=_get_id).start() - self.neutronclient.show_sfc_flow_classifier = mock.Mock( + self.network.get_sfc_flow_classifier = mock.Mock( return_value=self._flow_classifier ) # Get the command object to test @@ -263,8 +272,8 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2): self.namespace) def test_show_flow_classifier(self): - client = self.app.client_manager.neutronclient - mock_flow_classifier_show = client.show_sfc_flow_classifier + client = self.app.client_manager.network + mock_flow_classifier_show = client.get_sfc_flow_classifier arglist = [ self._flow_classifier_id, ] @@ -277,7 +286,6 @@ class TestShowSfcFlowClassifier(fakes.TestNeutronClientOSCV2): mock_flow_classifier_show.assert_called_once_with( self._flow_classifier_id) self.assertEqual(self.columns, columns) - self.assertEqual(self.data, data) class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2): @@ -324,9 +332,8 @@ class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestListSfcFlowClassifier, self).setUp() - mock.patch(get_id, new=_get_id).start() - self.neutronclient.list_sfc_flow_classifiers = mock.Mock( - return_value={'flow_classifiers': self._fc} + self.network.sfc_flow_classifiers = mock.Mock( + return_value=self._fc ) # Get the command object to test self.cmd = sfc_flow_classifier.ListSfcFlowClassifier(self.app, @@ -337,8 +344,7 @@ class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2): verifylist = [] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args) - fcs = self.neutronclient \ - .list_sfc_flow_classifiers()['flow_classifiers'] + fcs = self.network.sfc_flow_classifiers() fc = fcs[0] data = [ fc['id'], @@ -355,8 +361,7 @@ class TestListSfcFlowClassifier(fakes.TestNeutronClientOSCV2): def test_list_with_long_option(self): arglist = ['--long'] verifylist = [('long', True)] - fcs = self.neutronclient \ - .list_sfc_flow_classifiers()['flow_classifiers'] + fcs = self.network.sfc_flow_classifiers() fc = fcs[0] data = [ fc['id'], diff --git a/neutronclient/tests/unit/osc/v2/sfc/test_port_chain.py b/neutronclient/tests/unit/osc/v2/sfc/test_port_chain.py index da6455230..9b414c22c 100755 --- a/neutronclient/tests/unit/osc/v2/sfc/test_port_chain.py +++ b/neutronclient/tests/unit/osc/v2/sfc/test_port_chain.py @@ -16,6 +16,7 @@ from unittest import mock from osc_lib import exceptions +import testtools from neutronclient.osc.v2.sfc import sfc_port_chain from neutronclient.tests.unit.osc.v2.sfc import fakes @@ -29,8 +30,7 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2): # The new port_chain created _port_chain = fakes.FakeSfcPortChain.create_port_chain() - columns = ('Chain ID', - 'Chain Parameters', + columns = ('Chain Parameters', 'Description', 'Flow Classifiers', 'ID', @@ -40,7 +40,6 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2): def get_data(self): return ( - self._port_chain['chain_id'], self._port_chain['chain_parameters'], self._port_chain['description'], self._port_chain['flow_classifiers'], @@ -52,11 +51,8 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestCreateSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.neutronclient.create_sfc_port_chain = mock.Mock( - return_value={'port_chain': self._port_chain}) + self.network.create_sfc_port_chain = mock.Mock( + return_value=self._port_chain) self.data = self.get_data() # Get the command object to test @@ -77,11 +73,12 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_port_chain.assert_called_once_with({ - 'port_chain': { + self.network.create_sfc_port_chain.assert_called_once_with( + **{ 'name': self._port_chain['name'], - 'port_pair_groups': [self._port_chain['port_pair_groups']]} - }) + 'port_pair_groups': [self._port_chain['port_pair_groups']] + } + ) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -107,15 +104,15 @@ class TestCreateSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_port_chain.assert_called_once_with({ - 'port_chain': { + self.network.create_sfc_port_chain.assert_called_once_with( + **{ 'name': self._port_chain['name'], 'port_pair_groups': [self._port_chain['port_pair_groups']], 'description': self._port_chain['description'], 'flow_classifiers': [self._port_chain['flow_classifiers']], 'chain_parameters': cp - } - }) + } + ) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -126,20 +123,17 @@ class TestDeleteSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestDeleteSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.neutronclient.delete_sfc_port_chain = mock.Mock(return_value=None) + self.network.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 + client = self.app.client_manager.network mock_port_chain_delete = client.delete_sfc_port_chain arglist = [ self._port_chain[0]['id'], ] verifylist = [ - ('port_chain', self._port_chain[0]['id']), + ('port_chain', [self._port_chain[0]['id']]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) @@ -147,13 +141,29 @@ class TestDeleteSfcPortChain(fakes.TestNeutronClientOSCV2): self._port_chain[0]['id']) self.assertIsNone(result) + def test_delete_multiple_port_chains_with_exception(self): + client = self.app.client_manager.network + target1 = self._port_chain[0]['id'] + arglist = [target1] + verifylist = [('port_chain', [target1])] + + client.find_sfc_port_chain.side_effect = [ + target1, exceptions.CommandError + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + msg = "1 of 2 port chain(s) failed to delete." + with testtools.ExpectedException(exceptions.CommandError) as e: + self.cmd.take_action(parsed_args) + self.assertEqual(msg, str(e)) + class TestListSfcPortChain(fakes.TestNeutronClientOSCV2): _port_chains = fakes.FakeSfcPortChain.create_port_chains(count=1) columns = ('ID', 'Name', 'Port Pair Groups', 'Flow Classifiers', - 'Chain Parameters', 'Chain ID') + 'Chain Parameters') columns_long = ('ID', 'Name', 'Port Pair Groups', 'Flow Classifiers', - 'Chain Parameters', 'Description', 'Chain ID', 'Project') + 'Chain Parameters', 'Description', 'Project') _port_chain = _port_chains[0] data = [ _port_chain['id'], @@ -161,13 +171,11 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2): _port_chain['port_pair_groups'], _port_chain['flow_classifiers'], _port_chain['chain_parameters'], - _port_chain['chain_id'] ] data_long = [ _port_chain['id'], _port_chain['name'], _port_chain['project_id'], - _port_chain['chain_id'], _port_chain['port_pair_groups'], _port_chain['flow_classifiers'], _port_chain['chain_parameters'], @@ -178,11 +186,8 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestListSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.neutronclient.list_sfc_port_chains = mock.Mock( - return_value={'port_chains': self._port_chains} + self.network.sfc_port_chains = mock.Mock( + return_value=self._port_chains ) # Get the command object to test self.cmd = sfc_port_chain.ListSfcPortChain(self.app, self.namespace) @@ -192,7 +197,7 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2): verifylist = [] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args)[0] - pcs = self.neutronclient.list_sfc_port_chains()['port_chains'] + pcs = self.network.sfc_port_chains() pc = pcs[0] data = [ pc['id'], @@ -200,7 +205,6 @@ class TestListSfcPortChain(fakes.TestNeutronClientOSCV2): pc['port_pair_groups'], pc['flow_classifiers'], pc['chain_parameters'], - pc['chain_id'] ] self.assertEqual(list(self.columns), columns) self.assertEqual(self.data, data) @@ -210,13 +214,12 @@ 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_sfc_port_chains()['port_chains'] + pcs = self.network.sfc_port_chains() pc = pcs[0] data = [ pc['id'], pc['name'], pc['project_id'], - pc['chain_id'], pc['port_pair_groups'], pc['flow_classifiers'], pc['chain_parameters'], @@ -237,14 +240,11 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestSetSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.mocked = self.neutronclient.update_sfc_port_chain + self.mocked = self.network.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 + client = self.app.client_manager.network mock_port_chain_update = client.update_sfc_port_chain arglist = [ self._port_chain_name, @@ -258,10 +258,9 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'port_chain': {'name': 'name_updated', - 'description': 'desc_updated'}} + attrs = {'name': 'name_updated', 'description': 'desc_updated'} mock_port_chain_update.assert_called_once_with(self._port_chain_name, - attrs) + **attrs) self.assertIsNone(result) def test_set_flow_classifiers(self): @@ -269,23 +268,12 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): 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 == 3: - self.neutronclient.find_resource.assert_called_with( - 'flow_classifier', fc2, cmd_resource='sfc_flow_classifier') - return {'id': args[1]} - - self.neutronclient.find_resource.side_effect = _mock_flow_classifier + self.network.find_sfc_port_chain = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'flow_classifiers': [self.pc_fc]} + ) + self.network.find_sfc_flow_classifier.side_effect = \ + lambda name_or_id, ignore_missing=False: {'id': name_or_id} arglist = [ target, '--flow-classifier', fc1, @@ -298,12 +286,11 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) expect = {'flow_classifiers': [self.pc_fc, fc1, fc2]} - self.mocked.assert_called_once_with(target, {self.res: expect}) - self.assertEqual(3, self.neutronclient.find_resource.call_count) + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_set_no_flow_classifier(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network mock_port_chain_update = client.update_sfc_port_chain arglist = [ self._port_chain_name, @@ -315,9 +302,9 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'port_chain': {'flow_classifiers': []}} + attrs = {'flow_classifiers': []} mock_port_chain_update.assert_called_once_with(self._port_chain_name, - attrs) + **attrs) self.assertIsNone(result) def test_set_port_pair_groups(self): @@ -326,25 +313,10 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): ppg1 = 'port_pair_group1' ppg2 = 'port_pair_group2' - 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 {'port_pair_groups': [self.pc_ppg]} - - if self.neutronclient.find_resource.call_count == 2: - self.neutronclient.find_resource.assert_called_with( - 'port_pair_group', ppg1, - cmd_resource='sfc_port_pair_group') - return {'id': args[1]} - - if self.neutronclient.find_resource.call_count == 3: - self.neutronclient.find_resource.assert_called_with( - 'port_pair_group', ppg2, - cmd_resource='sfc_port_pair_group') - return {'id': args[1]} - - self.neutronclient.find_resource.side_effect = _mock_flow_classifier + self.network.find_sfc_port_chain = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'port_pair_groups': [self.pc_ppg]} + ) arglist = [ target, '--port-pair-group', ppg1, @@ -357,22 +329,13 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) expect = {'port_pair_groups': [existing_ppg, ppg1, ppg2]} - self.mocked.assert_called_once_with(target, {self.res: expect}) - self.assertEqual(3, self.neutronclient.find_resource.call_count) + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_set_no_port_pair_group(self): target = self.resource['id'] ppg1 = 'port_pair_group1' - def _mock_port_pair_group(*args, **kwargs): - - if self.neutronclient.find_resource.call_count == 1: - self.neutronclient.find_resource.assert_called_with( - 'port_pair_group', ppg1, - cmd_resource='sfc_port_pair_group') - return {'id': args[1]} - self.neutronclient.find_resource.side_effect = _mock_port_pair_group arglist = [ target, '--no-port-pair-group', @@ -386,8 +349,7 @@ class TestSetSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) expect = {'port_pair_groups': [ppg1]} - self.mocked.assert_called_once_with(target, {self.res: expect}) - self.assertEqual(1, self.neutronclient.find_resource.call_count) + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_set_only_no_port_pair_group(self): @@ -409,7 +371,6 @@ class TestShowSfcPortChain(fakes.TestNeutronClientOSCV2): _pc = fakes.FakeSfcPortChain.create_port_chain() data = ( - _pc['chain_id'], _pc['chain_parameters'], _pc['description'], _pc['flow_classifiers'], @@ -418,10 +379,9 @@ class TestShowSfcPortChain(fakes.TestNeutronClientOSCV2): _pc['port_pair_groups'], _pc['project_id'] ) - _port_chain = {'port_chain': _pc} + _port_chain = _pc _port_chain_id = _pc['id'] - columns = ('Chain ID', - 'Chain Parameters', + columns = ('Chain Parameters', 'Description', 'Flow Classifiers', 'ID', @@ -431,18 +391,15 @@ class TestShowSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestShowSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.neutronclient.show_sfc_port_chain = mock.Mock( + self.network.get_sfc_port_chain = mock.Mock( return_value=self._port_chain ) # Get the command object to test self.cmd = sfc_port_chain.ShowSfcPortChain(self.app, self.namespace) def test_show_port_chain(self): - client = self.app.client_manager.neutronclient - mock_port_chain_show = client.show_sfc_port_chain + client = self.app.client_manager.network + mock_port_chain_show = client.get_sfc_port_chain arglist = [ self._port_chain_id, ] @@ -468,34 +425,21 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestUnsetSfcPortChain, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_chain._get_id', - new=_get_id).start() - self.neutronclient.update_sfc_port_chain = mock.Mock( + self.network.update_sfc_port_chain = mock.Mock( return_value=None) - self.mocked = self.neutronclient.update_sfc_port_chain + self.mocked = self.network.update_sfc_port_chain self.cmd = sfc_port_chain.UnsetSfcPortChain(self.app, self.namespace) def test_unset_port_pair_group(self): target = self.resource['id'] ppg1 = 'port_pair_group1' - def _mock_port_pair_group(*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 {'port_pair_groups': [self.pc_ppg]} - - if self.neutronclient.find_resource.call_count == 2: - self.neutronclient.find_resource.assert_called_with( - 'port_pair_group', ppg1, - cmd_resource='sfc_port_pair_group') - return {'id': args[1]} - if self.neutronclient.find_resource.call_count == 3: - self.neutronclient.find_resource.assert_called_with( - self.res, target, cmd_resource='sfc_port_chain') - return {'id': args[1]} - self.neutronclient.find_resource.side_effect = _mock_port_pair_group + self.network.find_sfc_port_chain = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'port_pair_groups': [self.pc_ppg]} + ) + self.network.find_sfc_port_pair_group.side_effect = \ + lambda name_or_id, ignore_missing=False: {'id': name_or_id} arglist = [ target, @@ -508,24 +452,18 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) expect = {'port_pair_groups': [self.pc_ppg]} - self.mocked.assert_called_once_with(target, {self.res: expect}) + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_unset_flow_classifier(self): target = self.resource['id'] fc1 = 'flow_classifier1' - - 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]} - self.neutronclient.find_resource.side_effect = _mock_flow_classifier + self.network.find_sfc_port_chain = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'flow_classifiers': [self.pc_fc]} + ) + self.network.find_sfc_flow_classifier.side_effect = \ + lambda name_or_id, ignore_missing=False: {'id': name_or_id} arglist = [ target, @@ -538,11 +476,11 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) expect = {'flow_classifiers': [self.pc_fc]} - self.mocked.assert_called_once_with(target, {self.res: expect}) + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_unset_all_flow_classifier(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network target = self.resource['id'] mock_port_chain_update = client.update_sfc_port_chain arglist = [ @@ -557,5 +495,5 @@ class TestUnsetSfcPortChain(fakes.TestNeutronClientOSCV2): result = self.cmd.take_action(parsed_args) expect = {'flow_classifiers': []} mock_port_chain_update.assert_called_once_with(target, - {self.res: expect}) + **expect) self.assertIsNone(result) diff --git a/neutronclient/tests/unit/osc/v2/sfc/test_port_pair.py b/neutronclient/tests/unit/osc/v2/sfc/test_port_pair.py index 8246e35d9..e0610a360 100755 --- a/neutronclient/tests/unit/osc/v2/sfc/test_port_pair.py +++ b/neutronclient/tests/unit/osc/v2/sfc/test_port_pair.py @@ -15,14 +15,13 @@ from unittest import mock +from osc_lib import exceptions +import testtools + from neutronclient.osc.v2.sfc import sfc_port_pair from neutronclient.tests.unit.osc.v2.sfc import fakes -def _get_id(client, id_or_name, resource): - return id_or_name - - class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2): # The new port_pair created _port_pair = fakes.FakeSfcPortPair.create_port_pair() @@ -48,10 +47,8 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestCreateSfcPortPair, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id', - new=_get_id).start() - self.neutronclient.create_sfc_port_pair = mock.Mock( - return_value={'port_pair': self._port_pair}) + self.network.create_sfc_port_pair = mock.Mock( + return_value=self._port_pair) self.data = self.get_data() # Get the command object to test @@ -71,12 +68,11 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - 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'], - } - }) + self.network.create_sfc_port_pair.assert_called_once_with( + **{'name': self._port_pair['name'], + 'ingress': self._port_pair['ingress'], + 'egress': self._port_pair['egress']} + ) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -106,16 +102,14 @@ class TestCreateSfcPortPair(fakes.TestNeutronClientOSCV2): correlation_param = None else: correlation_param = correlation - 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'], - 'description': self._port_pair['description'], - 'service_function_parameters': - {'correlation': correlation_param, 'weight': - '1'}, - } - }) + self.network.create_sfc_port_pair.assert_called_once_with( + **{'name': self._port_pair['name'], + 'ingress': self._port_pair['ingress'], + 'egress': self._port_pair['egress'], + 'description': self._port_pair['description'], + 'service_function_parameters': + {'correlation': correlation_param, 'weight': '1'}} + ) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -132,19 +126,17 @@ class TestDeleteSfcPortPair(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestDeleteSfcPortPair, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id', - new=_get_id).start() - self.neutronclient.delete_sfc_port_pair = mock.Mock(return_value=None) + self.network.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 + client = self.app.client_manager.network mock_port_pair_delete = client.delete_sfc_port_pair arglist = [ self._port_pair[0]['id'], ] verifylist = [ - ('port_pair', self._port_pair[0]['id']), + ('port_pair', [self._port_pair[0]['id']]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) @@ -152,6 +144,22 @@ class TestDeleteSfcPortPair(fakes.TestNeutronClientOSCV2): self._port_pair[0]['id']) self.assertIsNone(result) + def test_delete_multiple_port_pairs_with_exception(self): + client = self.app.client_manager.network + target1 = self._port_pair[0]['id'] + arglist = [target1] + verifylist = [('port_pair', [target1])] + + client.find_sfc_port_pair.side_effect = [ + target1, exceptions.CommandError + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + msg = "1 of 2 port pair(s) failed to delete." + with testtools.ExpectedException(exceptions.CommandError) as e: + self.cmd.take_action(parsed_args) + self.assertEqual(msg, str(e)) + class TestListSfcPortPair(fakes.TestNeutronClientOSCV2): _port_pairs = fakes.FakeSfcPortPair.create_port_pairs() @@ -179,11 +187,8 @@ class TestListSfcPortPair(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestListSfcPortPair, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id', - new=_get_id).start() - self.neutronclient.list_sfc_port_pairs = mock.Mock( - return_value={'port_pairs': self._port_pairs} - ) + self.network.sfc_port_pairs = mock.Mock( + return_value=self._port_pairs) # Get the command object to test self.cmd = sfc_port_pair.ListSfcPortPair(self.app, self.namespace) @@ -192,7 +197,7 @@ class TestListSfcPortPair(fakes.TestNeutronClientOSCV2): verifylist = [] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args)[0] - port_pairs = self.neutronclient.list_sfc_port_pairs()['port_pairs'] + port_pairs = self.network.sfc_port_pairs() port_pair = port_pairs[0] data = [ port_pair['id'], @@ -206,7 +211,7 @@ class TestListSfcPortPair(fakes.TestNeutronClientOSCV2): def test_list_with_long_option(self): arglist = ['--long'] verifylist = [('long', True)] - port_pairs = self.neutronclient.list_sfc_port_pairs()['port_pairs'] + port_pairs = self.network.sfc_port_pairs() port_pair = port_pairs[0] data = [ port_pair['id'], @@ -229,13 +234,11 @@ class TestSetSfcPortPair(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestSetSfcPortPair, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id', - new=_get_id).start() - self.neutronclient.update_sfc_port_pair = mock.Mock(return_value=None) + self.network.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 + client = self.app.client_manager.network mock_port_pair_update = client.update_sfc_port_pair arglist = [ self._port_pair_name, @@ -249,12 +252,12 @@ class TestSetSfcPortPair(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'port_pair': { + attrs = { 'name': 'name_updated', - 'description': 'desc_updated'} + 'description': 'desc_updated' } mock_port_pair_update.assert_called_once_with(self._port_pair_name, - attrs) + **attrs) self.assertIsNone(result) @@ -271,7 +274,7 @@ class TestShowSfcPortPair(fakes.TestNeutronClientOSCV2): _pp['project_id'], _pp['service_function_parameters'], ) - _port_pair = {'port_pair': _pp} + _port_pair = _pp _port_pair_id = _pp['id'] columns = ( 'Description', @@ -285,10 +288,8 @@ class TestShowSfcPortPair(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestShowSfcPortPair, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_port_pair._get_id', - new=_get_id).start() - self.neutronclient.show_sfc_port_pair = mock.Mock( + self.network.get_sfc_port_pair = mock.Mock( return_value=self._port_pair ) @@ -296,8 +297,8 @@ class TestShowSfcPortPair(fakes.TestNeutronClientOSCV2): self.cmd = sfc_port_pair.ShowSfcPortPair(self.app, self.namespace) def test_show_port_pair(self): - client = self.app.client_manager.neutronclient - mock_port_pair_show = client.show_sfc_port_pair + client = self.app.client_manager.network + mock_port_pair_show = client.get_sfc_port_pair arglist = [ self._port_pair_id, ] diff --git a/neutronclient/tests/unit/osc/v2/sfc/test_port_pair_group.py b/neutronclient/tests/unit/osc/v2/sfc/test_port_pair_group.py index 6db21a284..49f3fcf06 100755 --- a/neutronclient/tests/unit/osc/v2/sfc/test_port_pair_group.py +++ b/neutronclient/tests/unit/osc/v2/sfc/test_port_pair_group.py @@ -15,21 +15,19 @@ from unittest import mock +from osc_lib import exceptions +import testtools + from neutronclient.osc.v2.sfc import sfc_port_pair_group from neutronclient.tests.unit.osc.v2.sfc import fakes -def _get_id(client, id_or_name, resource): - return id_or_name - - class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): _port_pair_group = fakes.FakeSfcPortPairGroup.create_port_pair_group() columns = ('Description', 'ID', - 'Loadbalance ID', 'Name', 'Port Pair', 'Port Pair Group Parameters', @@ -40,7 +38,6 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): return ( ppg['description'], ppg['id'], - ppg['group_id'], ppg['name'], ppg['port_pairs'], ppg['port_pair_group_parameters'], @@ -50,12 +47,10 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestCreateSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.create_sfc_port_pair_group = mock.Mock( - return_value={'port_pair_group': self._port_pair_group}) + self.network.create_sfc_port_pair_group = mock.Mock( + return_value=self._port_pair_group) self.data = self.get_data(self._port_pair_group) + # Get the command object to test self.cmd = sfc_port_pair_group.CreateSfcPortPairGroup(self.app, self.namespace) @@ -72,12 +67,9 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - 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']] - } - }) + self.network.create_sfc_port_pair_group.assert_called_once_with( + **{'name': self._port_pair_group['name'], + 'port_pairs': [self._port_pair_group['port_pairs']]}) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -95,13 +87,11 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_port_pair_group.assert_called_once_with({ - 'port_pair_group': { + self.network.create_sfc_port_pair_group.assert_called_once_with( + **{ 'name': self._port_pair_group['name'], 'port_pairs': [self._port_pair_group['port_pairs']], - 'description': self._port_pair_group['description'], - } - }) + 'description': self._port_pair_group['description']}) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) @@ -127,22 +117,19 @@ class TestCreateSfcPortPairGroup(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_port_pair_group.assert_called_once_with({ - 'port_pair_group': { + self.network.create_sfc_port_pair_group.assert_called_once_with( + **{ 'name': self._port_pair_group['name'], 'port_pairs': [self._port_pair_group['port_pairs']], 'description': self._port_pair_group['description'], - 'tap_enabled': True - } - }) + 'tap_enabled': True}) self.assertEqual(self.columns, columns) self.assertEqual(expected_data, data) def _update_expected_response_data(self, data): # REVISIT(vks1) - This method can be common for other test functions. ppg = fakes.FakeSfcPortPairGroup.create_port_pair_group(data) - self.neutronclient.create_sfc_port_pair_group.return_value = { - 'port_pair_group': ppg} + self.network.create_sfc_port_pair_group.return_value = ppg return self.get_data(ppg) @@ -153,22 +140,19 @@ class TestDeleteSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestDeleteSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.delete_sfc_port_pair_group = mock.Mock( + self.network.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 + client = self.app.client_manager.network mock_port_pair_group_delete = client.delete_sfc_port_pair_group arglist = [ self._port_pair_group[0]['id'], ] verifylist = [ - ('port_pair_group', self._port_pair_group[0]['id']), + ('port_pair_group', [self._port_pair_group[0]['id']]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) @@ -176,13 +160,29 @@ class TestDeleteSfcPortPairGroup(fakes.TestNeutronClientOSCV2): self._port_pair_group[0]['id']) self.assertIsNone(result) + def test_delete_multiple_port_pair_groups_with_exception(self): + client = self.app.client_manager.network + target1 = 'target' + arglist = [target1] + verifylist = [('port_pair_group', [target1])] + + client.find_sfc_port_pair_group.side_effect = [ + target1, exceptions.CommandError + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + msg = "1 of 2 port pair group(s) failed to delete." + with testtools.ExpectedException(exceptions.CommandError) as e: + self.cmd.take_action(parsed_args) + self.assertEqual(msg, str(e)) + class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2): _ppgs = fakes.FakeSfcPortPairGroup.create_port_pair_groups(count=1) columns = ('ID', 'Name', 'Port Pair', 'Port Pair Group Parameters', 'Tap Enabled') columns_long = ('ID', 'Name', 'Port Pair', 'Port Pair Group Parameters', - 'Description', 'Loadbalance ID', 'Project', 'Tap Enabled') + 'Description', 'Project', 'Tap Enabled') _port_pair_group = _ppgs[0] data = [ _port_pair_group['id'], @@ -204,12 +204,9 @@ class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestListSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.list_sfc_port_pair_groups = mock.Mock( - return_value={'port_pair_groups': self._ppgs} + self.network.sfc_port_pair_groups = mock.Mock( + return_value=self._ppgs ) # Get the command object to test self.cmd = sfc_port_pair_group.ListSfcPortPairGroup(self.app, @@ -220,8 +217,7 @@ class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2): verifylist = [] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args)[0] - ppgs = self.neutronclient \ - .list_sfc_port_pair_groups()['port_pair_groups'] + ppgs = self.network.sfc_port_pair_groups() ppg = ppgs[0] data = [ ppg['id'], @@ -236,8 +232,7 @@ class TestListSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def test_list_with_long_option(self): arglist = ['--long'] verifylist = [('long', True)] - ppgs = self.neutronclient \ - .list_sfc_port_pair_groups()['port_pair_groups'] + ppgs = self.network.sfc_port_pair_groups() ppg = ppgs[0] data = [ ppg['id'], @@ -264,12 +259,9 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestSetSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.update_sfc_port_pair_group = mock.Mock( + self.network.update_sfc_port_pair_group = mock.Mock( return_value=None) - self.mocked = self.neutronclient.update_sfc_port_pair_group + self.mocked = self.network.update_sfc_port_pair_group self.cmd = sfc_port_pair_group.SetSfcPortPairGroup(self.app, self.namespace) @@ -278,24 +270,14 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): port_pair1 = 'additional_port1' port_pair2 = 'additional_port2' - def _mock_port_pair_group(*args, **kwargs): - - if self.neutronclient.find_resource.call_count == 1: - self.neutronclient.find_resource.assert_called_with( - 'port_pair', port_pair1, cmd_resource='sfc_port_pair') - return {'id': args[1]} - - if self.neutronclient.find_resource.call_count == 2: - self.neutronclient.find_resource.assert_called_with( - 'port_pair', port_pair2, cmd_resource='sfc_port_pair') - return {'id': args[1]} - - if self.neutronclient.find_resource.call_count == 3: - self.neutronclient.find_resource.assert_called_with( - self.res, target, cmd_resource='sfc_port_pair_group') - return {'port_pairs': [self.ppg_pp]} - - self.neutronclient.find_resource.side_effect = _mock_port_pair_group + self.network.find_sfc_port_pair = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_port_pair_group = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'port_pairs': self.ppg_pp} + ) arglist = [ target, @@ -308,13 +290,12 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - expect = {'port_pairs': sorted([self.ppg_pp, port_pair1, port_pair2])} - self.mocked.assert_called_once_with(target, {self.res: expect}) - self.assertEqual(3, self.neutronclient.find_resource.call_count) + expect = {'port_pairs': sorted([*self.ppg_pp, port_pair1, port_pair2])} + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_set_no_port_pair(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network mock_port_pair_group_update = client.update_sfc_port_pair_group arglist = [ self._port_pair_group_name, @@ -331,11 +312,10 @@ class TestSetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'port_pair_group': {'name': 'name_updated', - 'description': 'desc_updated', - 'port_pairs': []}} + attrs = {'name': 'name_updated', 'description': 'desc_updated', + 'port_pairs': []} mock_port_pair_group_update.assert_called_once_with( - self._port_pair_group_name, attrs) + self._port_pair_group_name, **attrs) self.assertIsNone(result) @@ -345,18 +325,16 @@ class TestShowSfcPortPairGroup(fakes.TestNeutronClientOSCV2): data = ( _ppg['description'], _ppg['id'], - _ppg['group_id'], _ppg['name'], _ppg['port_pairs'], _ppg['port_pair_group_parameters'], _ppg['project_id'], _ppg['tap_enabled']) - _port_pair_group = {'port_pair_group': _ppg} + _port_pair_group = _ppg _port_pair_group_id = _ppg['id'] columns = ( 'Description', 'ID', - 'Loadbalance ID', 'Name', 'Port Pair', 'Port Pair Group Parameters', @@ -366,19 +344,16 @@ class TestShowSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestShowSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.show_sfc_port_pair_group = mock.Mock( + self.network.get_sfc_port_pair_group = mock.Mock( return_value=self._port_pair_group ) self.cmd = sfc_port_pair_group.ShowSfcPortPairGroup(self.app, self.namespace) def test_show_port_pair_group(self): - client = self.app.client_manager.neutronclient - mock_port_pair_group_show = client.show_sfc_port_pair_group + client = self.app.client_manager.network + mock_port_pair_group_show = client.get_sfc_port_pair_group arglist = [ self._port_pair_group_id, ] @@ -404,12 +379,9 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestUnsetSfcPortPairGroup, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_port_pair_group._get_id', - new=_get_id).start() - self.neutronclient.update_sfc_port_pair_group = mock.Mock( + self.network.update_sfc_port_pair_group = mock.Mock( return_value=None) - self.mocked = self.neutronclient.update_sfc_port_pair_group + self.mocked = self.network.update_sfc_port_pair_group self.cmd = sfc_port_pair_group.UnsetSfcPortPairGroup( self.app, self.namespace) @@ -418,30 +390,14 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): port_pair1 = 'additional_port1' port_pair2 = 'additional_port2' - def _mock_port_pair(*args, **kwargs): - - if self.neutronclient.find_resource.call_count == 1: - self.neutronclient.find_resource.assert_called_with( - self.res, target, cmd_resource='sfc_port_pair_group') - return {'port_pairs': [self.ppg_pp]} - - if self.neutronclient.find_resource.call_count == 2: - self.neutronclient.find_resource.assert_called_with( - 'port_pair', port_pair1, cmd_resource='sfc_port_pair') - return {'id': args[1]} - - if self.neutronclient.find_resource.call_count == 3: - self.neutronclient.find_resource.assert_called_with( - 'port_pair', port_pair2, cmd_resource='sfc_port_pair') - return {'id': args[1]} - - if self.neutronclient.find_resource.call_count == 4: - self.neutronclient.find_resource.assert_called_with( - self.res, target, cmd_resource='sfc_port_pair_group') - return {'id': args[1]} - - self.neutronclient.find_resource.side_effect = _mock_port_pair - + self.network.find_sfc_port_pair = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id} + ) + self.network.find_sfc_port_pair_group = mock.Mock( + side_effect=lambda name_or_id, ignore_missing=False: + {'id': name_or_id, 'port_pairs': self.ppg_pp} + ) arglist = [ target, '--port-pair', port_pair1, @@ -453,12 +409,12 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - expect = {'port_pairs': sorted([self.ppg_pp])} - self.mocked.assert_called_once_with(target, {self.res: expect}) + expect = {'port_pairs': sorted([*self.ppg_pp])} + self.mocked.assert_called_once_with(target, **expect) self.assertIsNone(result) def test_unset_all_port_pair(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network mock_port_pair_group_update = client.update_sfc_port_pair_group arglist = [ self._port_pair_group_name, @@ -471,7 +427,7 @@ class TestUnsetSfcPortPairGroup(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'port_pair_group': {'port_pairs': []}} + attrs = {'port_pairs': []} mock_port_pair_group_update.assert_called_once_with( - self._port_pair_group_name, attrs) + self._port_pair_group_name, **attrs) self.assertIsNone(result) diff --git a/neutronclient/tests/unit/osc/v2/sfc/test_service_graph.py b/neutronclient/tests/unit/osc/v2/sfc/test_service_graph.py index a05ed2c72..50a02139d 100644 --- a/neutronclient/tests/unit/osc/v2/sfc/test_service_graph.py +++ b/neutronclient/tests/unit/osc/v2/sfc/test_service_graph.py @@ -11,19 +11,17 @@ # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # License for the specific language governing permissions and limitations # under the License. + from unittest import mock from osc_lib import exceptions from osc_lib.tests import utils as tests_utils +import testtools from neutronclient.osc.v2.sfc import sfc_service_graph from neutronclient.tests.unit.osc.v2.sfc import fakes -def _get_id(client, id_or_name, resource): - return id_or_name - - class TestListSfcServiceGraph(fakes.TestNeutronClientOSCV2): _service_graphs = fakes.FakeSfcServiceGraph.create_sfc_service_graphs( count=1) @@ -47,11 +45,8 @@ class TestListSfcServiceGraph(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestListSfcServiceGraph, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_service_graph._get_id', - new=_get_id).start() - self.neutronclient.list_sfc_service_graphs = mock.Mock( - return_value={'service_graphs': self._service_graphs} + self.network.sfc_service_graphs = mock.Mock( + return_value=self._service_graphs ) # Get the command object to test self.cmd = sfc_service_graph.ListSfcServiceGraph( @@ -62,7 +57,7 @@ class TestListSfcServiceGraph(fakes.TestNeutronClientOSCV2): verifylist = [] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args)[0] - sgs = self.neutronclient.list_sfc_service_graphs()['service_graphs'] + sgs = self.network.sfc_service_graphs() sg = sgs[0] data = [ sg['id'], @@ -77,7 +72,7 @@ class TestListSfcServiceGraph(fakes.TestNeutronClientOSCV2): verifylist = [('long', True)] parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns = self.cmd.take_action(parsed_args)[0] - sgs = self.neutronclient.list_sfc_service_graphs()['service_graphs'] + sgs = self.network.sfc_service_graphs() sg = sgs[0] data = [ sg['id'], @@ -107,10 +102,8 @@ class TestCreateSfcServiceGraph(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestCreateSfcServiceGraph, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_service_graph._get_id', - new=_get_id).start() - self.neutronclient.create_sfc_service_graph = mock.Mock( - return_value={'service_graph': self._service_graph}) + self.network.create_sfc_service_graph = mock.Mock( + return_value=self._service_graph) self.data = self.get_data() self.cmd = sfc_service_graph.CreateSfcServiceGraph( self.app, self.namespace) @@ -145,12 +138,10 @@ class TestCreateSfcServiceGraph(fakes.TestNeutronClientOSCV2): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = (self.cmd.take_action(parsed_args)) - self.neutronclient.create_sfc_service_graph.assert_called_once_with({ - 'service_graph': { + self.network.create_sfc_service_graph.assert_called_once_with(**{ 'description': self._service_graph['description'], 'name': self._service_graph['name'], 'port_chains': pcs - } }) self.assertEqual(self.columns_long, columns) self.assertEqual(self.data, data) @@ -227,19 +218,19 @@ class TestDeleteSfcServiceGraph(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestDeleteSfcServiceGraph, self).setUp() - self.neutronclient.delete_sfc_service_graph = mock.Mock( + self.network.delete_sfc_service_graph = mock.Mock( return_value=None) self.cmd = sfc_service_graph.DeleteSfcServiceGraph( self.app, self.namespace) def test_delete_sfc_service_graph(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network mock_service_graph_delete = client.delete_sfc_service_graph arglist = [ self._service_graph[0]['id'], ] verifylist = [ - ('service_graph', self._service_graph[0]['id']), + ('service_graph', [self._service_graph[0]['id']]), ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) @@ -247,6 +238,22 @@ class TestDeleteSfcServiceGraph(fakes.TestNeutronClientOSCV2): self._service_graph[0]['id']) self.assertIsNone(result) + def test_delete_multiple_service_graphs_with_exception(self): + client = self.app.client_manager.network + target = self._service_graph[0]['id'] + arglist = [target] + verifylist = [('service_graph', [target])] + + client.find_sfc_service_graph.side_effect = [ + target, exceptions.CommandError + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + msg = "1 of 2 service graph(s) failed to delete." + with testtools.ExpectedException(exceptions.CommandError) as e: + self.cmd.take_action(parsed_args) + self.assertEqual(msg, str(e)) + class TestShowSfcServiceGraph(fakes.TestNeutronClientOSCV2): @@ -266,15 +273,12 @@ class TestShowSfcServiceGraph(fakes.TestNeutronClientOSCV2): _sg['project_id'] ) - _service_graph = {'service_graph': _sg} + _service_graph = _sg _service_graph_id = _sg['id'] def setUp(self): super(TestShowSfcServiceGraph, self).setUp() - mock.patch( - 'neutronclient.osc.v2.sfc.sfc_service_graph._get_id', - new=_get_id).start() - self.neutronclient.show_sfc_service_graph = mock.Mock( + self.network.get_sfc_service_graph = mock.Mock( return_value=self._service_graph ) # Get the command object to test @@ -282,8 +286,8 @@ class TestShowSfcServiceGraph(fakes.TestNeutronClientOSCV2): self.app, self.namespace) def test_service_graph_show(self): - client = self.app.client_manager.neutronclient - mock_service_graph_show = client.show_sfc_service_graph + client = self.app.client_manager.network + mock_service_graph_show = client.get_sfc_service_graph arglist = [ self._service_graph_id, ] @@ -305,15 +309,13 @@ class TestSetSfcServiceGraph(fakes.TestNeutronClientOSCV2): def setUp(self): super(TestSetSfcServiceGraph, self).setUp() - mock.patch('neutronclient.osc.v2.sfc.sfc_service_graph._get_id', - new=_get_id).start() - self.neutronclient.update_sfc_service_graph = mock.Mock( + self.network.update_sfc_service_graph = mock.Mock( return_value=None) self.cmd = sfc_service_graph.SetSfcServiceGraph( self.app, self.namespace) def test_set_service_graph(self): - client = self.app.client_manager.neutronclient + client = self.app.client_manager.network mock_service_graph_update = client.update_sfc_service_graph arglist = [ self._service_graph_name, @@ -327,10 +329,10 @@ class TestSetSfcServiceGraph(fakes.TestNeutronClientOSCV2): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - attrs = {'service_graph': { + attrs = { 'name': 'name_updated', - 'description': 'desc_updated'} + 'description': 'desc_updated' } mock_service_graph_update.assert_called_once_with( - self._service_graph_name, attrs) + self._service_graph_name, **attrs) self.assertIsNone(result)