From c772a1dbc7ace58d099570d41a889adf851c8ba8 Mon Sep 17 00:00:00 2001 From: Ulas Kozat Date: Mon, 28 Dec 2015 16:05:13 -0800 Subject: Added networking-sfc from openstack project with merge date Dec 23 2015 Added patch 13 for subject "add missing db migration files" Change-Id: Id51a160335a14870c1dd816a44baf9b1958b9ac6 --- networking_sfc/tests/unit/cli/__init__.py | 0 .../tests/unit/cli/test_flow_classifier.py | 182 ++++++++++++++++++++ networking_sfc/tests/unit/cli/test_port_chain.py | 186 +++++++++++++++++++++ networking_sfc/tests/unit/cli/test_port_pair.py | 160 ++++++++++++++++++ .../tests/unit/cli/test_port_pair_group.py | 144 ++++++++++++++++ 5 files changed, 672 insertions(+) create mode 100644 networking_sfc/tests/unit/cli/__init__.py create mode 100644 networking_sfc/tests/unit/cli/test_flow_classifier.py create mode 100644 networking_sfc/tests/unit/cli/test_port_chain.py create mode 100644 networking_sfc/tests/unit/cli/test_port_pair.py create mode 100644 networking_sfc/tests/unit/cli/test_port_pair_group.py (limited to 'networking_sfc/tests/unit/cli') diff --git a/networking_sfc/tests/unit/cli/__init__.py b/networking_sfc/tests/unit/cli/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/networking_sfc/tests/unit/cli/test_flow_classifier.py b/networking_sfc/tests/unit/cli/test_flow_classifier.py new file mode 100644 index 0000000..37c2142 --- /dev/null +++ b/networking_sfc/tests/unit/cli/test_flow_classifier.py @@ -0,0 +1,182 @@ +# Copyright 2015 Huawei Technologies India Pvt. Ltd. +# All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +import sys +import uuid + +import mock + +from neutronclient import shell +from neutronclient.tests.unit import test_cli20 + +from networking_sfc.cli import flow_classifier as fc + +source_port_UUID = str(uuid.uuid4()) +destination_port_UUID = str(uuid.uuid4()) + + +class CLITestV20FCExtensionJSON(test_cli20.CLITestV20Base): + def setUp(self): + self._mock_extension_loading() + super(CLITestV20FCExtensionJSON, self).setUp(plurals={}) + self.register_non_admin_status_resource('flow_classifier') + + def _create_patch(self, name, func=None): + patcher = mock.patch(name) + thing = patcher.start() + self.addCleanup(patcher.stop) + return thing + + def _mock_extension_loading(self): + ext_pkg = 'neutronclient.common.extension' + flow_classifier = self._create_patch(ext_pkg + + '._discover_via_entry_points') + flow_classifier.return_value = [("flow_classifier", fc)] + return flow_classifier + + def test_ext_cmd_loaded(self): + shell.NeutronShell('2.0') + ext_cmd = {'flow-classifier-list': fc.FlowClassifierList, + 'flow-classifier-create': fc.FlowClassifierCreate, + 'flow-classifier-update': fc.FlowClassifierUpdate, + 'flow-classifier-delete': fc.FlowClassifierDelete, + 'flow-classifier-show': fc.FlowClassifierShow} + self.assertDictContainsSubset(ext_cmd, shell.COMMANDS['2.0']) + + def test_create_flow_classifier_with_mandatory_params(self): + """create flow-classifier: flow1.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierCreate(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + name = 'flow1' + ethertype = 'IPv4' + args = [name, '--ethertype', ethertype] + position_names = ['name', 'ethertype'] + position_values = [name, ethertype] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_flow_classifier_with_all_params(self): + """create flow-classifier: flow1.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierCreate(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + name = 'flow1' + protocol_name = 'TCP' + ethertype = 'IPv4' + source_port = '0:65535' + source_port_min = 0 + source_port_max = 65535 + destination_port = '1:65534' + destination_port_min = 1 + destination_port_max = 65534 + source_ip = '192.168.1.0/24' + destination_ip = '192.168.2.0/24' + logical_source_port = '4a334cd4-fe9c-4fae-af4b-321c5e2eb051' + logical_destination_port = '1278dcd4-459f-62ed-754b-87fc5e4a6751' + description = 'my-desc' + l7_param = "url=my_url" + l7_param_expected = {"url": "my_url"} + args = [name, + '--protocol', protocol_name, + '--ethertype', ethertype, + '--source-port', source_port, + '--destination-port', destination_port, + '--source-ip-prefix', source_ip, + '--destination-ip-prefix', destination_ip, + '--logical-source-port', logical_source_port, + '--logical-destination-port', logical_destination_port, + '--description', description, + '--l7-parameters', l7_param] + position_names = ['name', 'protocol', 'ethertype', + 'source_port_range_min', 'source_port_range_max', + 'destination_port_range_min', + 'destination_port_range_max', + 'source_ip_prefix', 'destination_ip_prefix', + 'logical_source_port', 'logical_destination_port', + 'description', 'l7_parameters'] + position_values = [name, protocol_name, ethertype, + source_port_min, source_port_max, + destination_port_min, destination_port_max, + source_ip, destination_ip, logical_source_port, + logical_destination_port, description, + l7_param_expected] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_list_flow_classifier(self): + """List available flow-classifiers.""" + resources = "flow_classifiers" + cmd = fc.FlowClassifierList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, True) + + def test_list_flow_classifier_sort(self): + """flow_classifier-list --sort-key name --sort-key id --sort-key asc + + --sort-key desc + """ + resources = "flow_classifiers" + cmd = fc.FlowClassifierList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, + sort_key=["name", "id"], + sort_dir=["asc", "desc"]) + + def test_list_flow_classifier_limit(self): + """flow-classifier-list -P.""" + resources = "flow_classifiers" + cmd = fc.FlowClassifierList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, page_size=1000) + + def test_show_flow_classifier_id(self): + """flow-classifier-show test_id.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierShow(test_cli20.MyApp(sys.stdout), None) + args = ['--fields', 'id', self.test_id] + self._test_show_resource(resource, cmd, self.test_id, args, ['id']) + + def test_show_flow_classifier_id_name(self): + """flow-classifier-show .""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierShow(test_cli20.MyApp(sys.stdout), None) + args = ['--fields', 'id', '--fields', 'name', self.test_id] + self._test_show_resource(resource, cmd, self.test_id, + args, ['id', 'name']) + + def test_update_flow_classifier_description(self): + """flow-classifier-update myid --name newname.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierUpdate(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid, '--description', 'flow_classifier1', '--description', + 'flow_classifier2'] + updatefields = {'description': 'flow_classifier2'} + self._test_update_resource(resource, cmd, myid, args, updatefields) + + def test_update_flow_classifier_name(self): + """flow-classifier-update myid --protocol any.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierUpdate(test_cli20.MyApp(sys.stdout), None) + self._test_update_resource(resource, cmd, 'myid', + ['myid', '--name', 'myname'], + {'name': 'myname'}) + + def test_delete_flow_classifer(self): + """flow-classifier-delete my-id.""" + resource = 'flow_classifier' + cmd = fc.FlowClassifierDelete(test_cli20.MyApp(sys.stdout), None) + my_id = 'myid1' + args = [my_id] + self._test_delete_resource(resource, cmd, my_id, args) diff --git a/networking_sfc/tests/unit/cli/test_port_chain.py b/networking_sfc/tests/unit/cli/test_port_chain.py new file mode 100644 index 0000000..0d834bd --- /dev/null +++ b/networking_sfc/tests/unit/cli/test_port_chain.py @@ -0,0 +1,186 @@ +# Copyright 2015 Huawei Technologies India Pvt. Ltd. +# All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +import sys +import uuid + +import mock + +from neutronclient import shell +from neutronclient.tests.unit import test_cli20 + +from networking_sfc.cli import port_chain as pc + +FAKE_port_pair_group1_UUID = str(uuid.uuid4()) +FAKE_port_pair_group2_UUID = str(uuid.uuid4()) +FAKE_FC1_UUID = str(uuid.uuid4()) +FAKE_FC2_UUID = str(uuid.uuid4()) +FAKE_PARAM1_UUID = str(uuid.uuid4()) +FAKE_PARAM2_UUID = str(uuid.uuid4()) + + +class CLITestV20PortChainExtensionJSON(test_cli20.CLITestV20Base): + def setUp(self): + self._mock_extension_loading() + super(CLITestV20PortChainExtensionJSON, self).setUp() + self.register_non_admin_status_resource('port_chain') + + def _create_patch(self, name, func=None): + patcher = mock.patch(name) + thing = patcher.start() + self.addCleanup(patcher.stop) + return thing + + def _mock_extension_loading(self): + ext_pkg = 'neutronclient.common.extension' + port_chain = self._create_patch(ext_pkg + + '._discover_via_entry_points') + port_chain.return_value = [("port_chain", pc)] + return port_chain + + def test_ext_cmd_loaded(self): + shell.NeutronShell('2.0') + ext_cmd = {'port-chain-list': pc.PortChainList, + 'port-chain-create': pc.PortChainCreate, + 'port-chain-update': pc.PortChainUpdate, + 'port-chain-delete': pc.PortChainDelete, + 'port-chain-show': pc.PortChainShow} + self.assertDictContainsSubset(ext_cmd, shell.COMMANDS['2.0']) + + def test_create_port_chain_with_mandatory_param(self): + """Create port_chain: myname.""" + resource = 'port_chain' + cmd = pc.PortChainCreate(test_cli20.MyApp(sys.stdout), + None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair-group', FAKE_port_pair_group1_UUID] + position_names = ['name', 'port_pair_groups'] + position_values = [name, [FAKE_port_pair_group1_UUID]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_chain_with_multiple_port_pair_group(self): + """Create port_chain: myname.""" + resource = 'port_chain' + cmd = pc.PortChainCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair-group', FAKE_port_pair_group1_UUID, + '--port-pair-group', FAKE_port_pair_group2_UUID] + position_names = ['name', 'port_pair_groups'] + position_values = [name, [FAKE_port_pair_group1_UUID, + FAKE_port_pair_group2_UUID]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_chain_with_all_params(self): + """Create port_chain: myname.""" + resource = 'port_chain' + cmd = pc.PortChainCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + desc = 'check port chain cli' + chain_parameter = "correlation=mpls" + chain_parameter_expected = {"correlation": "mpls"} + args = [name, '--description', desc, '--port-pair-group', + FAKE_port_pair_group1_UUID, '--flow-classifier', + FAKE_FC1_UUID, '--chain-parameters', chain_parameter] + position_names = ['name', 'description', 'port_pair_groups', + 'flow_classifiers', 'chain_parameters'] + position_values = [name, desc, [FAKE_port_pair_group1_UUID], + [FAKE_FC1_UUID], chain_parameter_expected] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_chain_with_single_classifier(self): + """Create port_chain: myname.""" + resource = 'port_chain' + cmd = pc.PortChainCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair-group', FAKE_port_pair_group1_UUID, + '--flow-classifier', FAKE_FC1_UUID] + position_names = ['name', 'port_pair_groups', 'flow_classifiers'] + position_values = [name, [FAKE_port_pair_group1_UUID], [FAKE_FC1_UUID]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_chain_with_multiple_classifiers(self): + """Create port_chain: myname.""" + resource = 'port_chain' + cmd = pc.PortChainCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair-group', FAKE_port_pair_group1_UUID, + '--flow-classifier', FAKE_FC1_UUID, + '--flow-classifier', FAKE_FC2_UUID] + position_names = ['name', 'port_pair_groups', 'flow_classifiers'] + position_values = [name, [FAKE_port_pair_group1_UUID], [FAKE_FC1_UUID, + FAKE_FC2_UUID]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_update_port_chain(self): + """Update port_chain: myid --name myname.""" + resource = 'port_chain' + cmd = pc.PortChainUpdate(test_cli20.MyApp(sys.stdout), None) + self._test_update_resource(resource, cmd, 'myid', + ['myid', '--name', 'myname'], + {'name': 'myname'}) + + def test_update_port_chain_with_no_flow_classifier(self): + """Update port_chain: myid --name myname --no-flow-classifier None.""" + resource = 'port_chain' + cmd = pc.PortChainUpdate(test_cli20.MyApp(sys.stdout), None) + self._test_update_resource(resource, cmd, 'myid', + ['myid', '--name', 'myname', + '--no-flow-classifier'], + {'name': 'myname', + 'flow_classifiers': []}) + + def test_delete_port_chain(self): + """Delete port-chain: myid.""" + resource = 'port_chain' + cmd = pc.PortChainDelete(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid] + self._test_delete_resource(resource, cmd, myid, args) + + def test_list_port_chain(self): + """List port_chain.""" + resources = 'port_chains' + cmd = pc.PortChainList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, True) + + def test_list_port_chains_sort(self): + """List port_chains: --sort-key name --sort-key id --sort-key asc + + --sort-key desc + """ + resources = "port_chains" + cmd = pc.PortChainList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, + sort_key=["name", "id"], + sort_dir=["asc", "desc"]) + + def test_show_port_chain(self): + """Show port-chain: --fields id --fields name myid.""" + resource = 'port_chain' + cmd = pc.PortChainShow(test_cli20.MyApp(sys.stdout), None) + args = ['--fields', 'id', '--fields', 'name', self.test_id] + self._test_show_resource(resource, cmd, self.test_id, + args, ['id', 'name']) diff --git a/networking_sfc/tests/unit/cli/test_port_pair.py b/networking_sfc/tests/unit/cli/test_port_pair.py new file mode 100644 index 0000000..9a302e2 --- /dev/null +++ b/networking_sfc/tests/unit/cli/test_port_pair.py @@ -0,0 +1,160 @@ +# Copyright 2015 Huawei Technologies India Pvt. Ltd. +# All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +import sys +import uuid + +import mock + +from neutronclient import shell +from neutronclient.tests.unit import test_cli20 + +from networking_sfc.cli import port_pair as pp + +ingress_port_UUID = str(uuid.uuid4()) +egress_port_UUID = str(uuid.uuid4()) + + +class CLITestV20PortPairExtensionJSON(test_cli20.CLITestV20Base): + def setUp(self): + self._mock_extension_loading() + super(CLITestV20PortPairExtensionJSON, self).setUp() + self.register_non_admin_status_resource('port_pair') + + def _create_patch(self, name, func=None): + patcher = mock.patch(name) + thing = patcher.start() + self.addCleanup(patcher.stop) + return thing + + def _mock_extension_loading(self): + ext_pkg = 'neutronclient.common.extension' + port_pair = self._create_patch(ext_pkg + + '._discover_via_entry_points') + port_pair.return_value = [("port_pair", pp)] + return port_pair + + def test_ext_cmd_loaded(self): + shell.NeutronShell('2.0') + ext_cmd = {'port-pair-list': pp.PortPairList, + 'port-pair-create': pp.PortPairCreate, + 'port-pair-update': pp.PortPairUpdate, + 'port-pair-delete': pp.PortPairDelete, + 'port-pair-show': pp.PortPairShow} + self.assertDictContainsSubset(ext_cmd, shell.COMMANDS['2.0']) + + def test_create_port_pair_with_mandatory_param(self): + """Create port_pair: myname.""" + resource = 'port_pair' + cmd = pp.PortPairCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--ingress', ingress_port_UUID, + '--egress', egress_port_UUID] + position_names = ['name', 'ingress', 'egress'] + position_values = [name, ingress_port_UUID, egress_port_UUID] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_group_with_bidirectional_port(self): + """Create port_pair: myname with bidirectional port.""" + resource = 'port_pair' + cmd = pp.PortPairCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--ingress', ingress_port_UUID, + '--egress', ingress_port_UUID] + position_names = ['name', 'ingress', 'egress'] + position_values = [name, ingress_port_UUID, ingress_port_UUID] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_pair_with_all_param(self): + """Create port_pair: myname with all parameter""" + resource = 'port_pair' + cmd = pp.PortPairCreate(test_cli20.MyApp(sys.stdout), + None) + name = 'myname' + myid = 'myid' + desc = "my_port_pair" + service_fn_param = 'correlation=None' + service_fn_param_exp = {"correlation": "None"} + args = [name, '--ingress', ingress_port_UUID, + '--egress', egress_port_UUID, '--description', desc, + '--service-function-parameters', service_fn_param] + position_names = ['name', 'ingress', 'egress', 'description', + 'service_function_parameters'] + position_values = [name, ingress_port_UUID, egress_port_UUID, desc, + service_fn_param_exp] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_update_port_pair_description(self): + """Update port_pair: myid --name myname.""" + resource = 'port_pair' + desc1 = "My_New_Port_Pair" + cmd = pp.PortPairUpdate(test_cli20.MyApp(sys.stdout), None) + self._test_update_resource(resource, cmd, 'myid', + ['myid', '--description', desc1], + {'description': desc1}) + + def test_update_port_pair_name(self): + """Update port_pair: myid --name myname.""" + resource = 'port_pair' + my_name = "My_New_Port_Pair" + cmd = pp.PortPairUpdate(test_cli20.MyApp(sys.stdout), None) + self._test_update_resource(resource, cmd, 'myid', + ['myid', '--name', my_name], + {'name': my_name}) + + def test_delete_port_pair(self): + """Delete port-pair: myid.""" + resource = 'port_pair' + cmd = pp.PortPairDelete(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid] + self._test_delete_resource(resource, cmd, myid, args) + + def test_list_port_pair(self): + """List port_pairs.""" + resources = 'port_pairs' + cmd = pp.PortPairList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, True) + + def test_list_port_pair_limit(self): + """size (1000) limited list: port-pair -P.""" + resources = "port_pairs" + cmd = pp.PortPairList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, page_size=1000) + + def test_list_port_pairs_sort(self): + """List port_pairs: --sort-key name --sort-key id --sort-key asc + + --sort-key desc + """ + resources = "port_pairs" + cmd = pp.PortPairList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, + sort_key=["name", "id"], + sort_dir=["asc", "desc"]) + + def test_show_port_pair(self): + """Show port-pairs: --fields id --fields name myid.""" + resource = 'port_pair' + cmd = pp.PortPairShow(test_cli20.MyApp(sys.stdout), None) + args = ['--fields', 'id', '--fields', 'name', self.test_id] + self._test_show_resource(resource, cmd, self.test_id, + args, ['id', 'name']) diff --git a/networking_sfc/tests/unit/cli/test_port_pair_group.py b/networking_sfc/tests/unit/cli/test_port_pair_group.py new file mode 100644 index 0000000..f610ef0 --- /dev/null +++ b/networking_sfc/tests/unit/cli/test_port_pair_group.py @@ -0,0 +1,144 @@ +# Copyright 2015 Huawei Technologies India Pvt. Ltd. +# All Rights Reserved +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# + +import sys +import uuid + +import mock + +from neutronclient import shell +from neutronclient.tests.unit import test_cli20 + +from networking_sfc.cli import port_pair_group as pg + +pp1 = str(uuid.uuid4()) +pp2 = str(uuid.uuid4()) +pp3 = str(uuid.uuid4()) +pp4 = str(uuid.uuid4()) + + +class CLITestV20PortGroupExtensionJSON(test_cli20.CLITestV20Base): + def setUp(self): + self._mock_extension_loading() + super(CLITestV20PortGroupExtensionJSON, self).setUp() + self.register_non_admin_status_resource('port_pair_group') + + def _create_patch(self, name, func=None): + patcher = mock.patch(name) + thing = patcher.start() + self.addCleanup(patcher.stop) + return thing + + def _mock_extension_loading(self): + ext_pkg = 'neutronclient.common.extension' + port_pair_group = self._create_patch(ext_pkg + + '._discover_via_entry_points') + port_pair_group.return_value = [("port_pair_group", pg)] + return port_pair_group + + def test_ext_cmd_loaded(self): + shell.NeutronShell('2.0') + ext_cmd = {'port-pair-group-list': pg.PortPairGroupList, + 'port-pair-group-create': pg.PortPairGroupCreate, + 'port-pair-group-update': pg.PortPairGroupUpdate, + 'port-pair-group-delete': pg.PortPairGroupDelete, + 'port-pair-group-show': pg.PortPairGroupShow} + self.assertDictContainsSubset(ext_cmd, shell.COMMANDS['2.0']) + + def test_create_port_pair_group_with_mandatory_args(self): + """Create port_pair_group: myname.""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair', pp1] + position_names = ['name', 'port_pairs'] + position_values = [name, [pp1]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_create_port_pair_group_with_ingress_egress_port_group(self): + """Create port_pair_group: myname with multiple port pairs""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupCreate(test_cli20.MyApp(sys.stdout), None) + name = 'myname' + myid = 'myid' + args = [name, '--port-pair', pp1, '--port-pair', pp2] + position_names = ['name', 'port_pairs'] + position_values = [name, [pp1, pp2]] + self._test_create_resource(resource, cmd, name, myid, args, + position_names, position_values) + + def test_delete_port_pair_group(self): + """Delete port_pair_group: myid.""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupDelete(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid] + self._test_delete_resource(resource, cmd, myid, args) + + def test_update_port_group_only_port_pair(self): + """Update port_pair_group""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupUpdate(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid, '--port-pair', pp1, + '--port-pair', pp2] + updatefields = {'port_pairs': [pp1, pp2]} + self._test_update_resource(resource, cmd, myid, args, updatefields) + + def test_update_port_group_with_all_desc(self): + """Update port_pair_group and description""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupUpdate(test_cli20.MyApp(sys.stdout), None) + myid = 'myid' + args = [myid, '--port-pair', pp1, + '--port-pair', pp2, '--description', 'my_port_group', + '--description', 'my_port_pair_group'] + updatefields = {'port_pairs': [pp1, pp2], + 'description': 'my_port_pair_group'} + self._test_update_resource(resource, cmd, myid, args, updatefields) + + def test_list_port_pair_group(self): + """List port_pair_group.""" + resources = 'port_pair_groups' + cmd = pg.PortPairGroupList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, True) + + def test_list_port_pair_group_limit(self): + """size (1000) limited list: port-pair-group -P.""" + resources = "port_pair_groups" + cmd = pg.PortPairGroupList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, page_size=1000) + + def test_list_port_group_sort(self): + """List port_pair_group: --sort-key name --sort-key id --sort-key asc + + --sort-key desc + """ + resources = "port_pair_groups" + cmd = pg.PortPairGroupList(test_cli20.MyApp(sys.stdout), None) + self._test_list_resources(resources, cmd, + sort_key=["name", "id"], + sort_dir=["asc", "desc"]) + + def test_show_port_group(self): + """Show port-chain: --fields id --fields name myid.""" + resource = 'port_pair_group' + cmd = pg.PortPairGroupShow(test_cli20.MyApp(sys.stdout), None) + args = ['--fields', 'id', '--fields', 'name', self.test_id] + self._test_show_resource(resource, cmd, self.test_id, + args, ['id', 'name']) -- cgit 1.2.3-korg