aboutsummaryrefslogtreecommitdiffstats
path: root/networking_sfc/tests/unit/cli
diff options
context:
space:
mode:
Diffstat (limited to 'networking_sfc/tests/unit/cli')
-rw-r--r--networking_sfc/tests/unit/cli/__init__.py0
-rw-r--r--networking_sfc/tests/unit/cli/test_flow_classifier.py182
-rw-r--r--networking_sfc/tests/unit/cli/test_port_chain.py186
-rw-r--r--networking_sfc/tests/unit/cli/test_port_pair.py160
-rw-r--r--networking_sfc/tests/unit/cli/test_port_pair_group.py144
5 files changed, 672 insertions, 0 deletions
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
--- /dev/null
+++ b/networking_sfc/tests/unit/cli/__init__.py
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'])