diff options
Diffstat (limited to 'openstackclient/tests/functional/network')
8 files changed, 979 insertions, 96 deletions
diff --git a/openstackclient/tests/functional/network/v2/test_address_scope.py b/openstackclient/tests/functional/network/v2/test_address_scope.py index ef4b5756..75f84344 100644 --- a/openstackclient/tests/functional/network/v2/test_address_scope.py +++ b/openstackclient/tests/functional/network/v2/test_address_scope.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import re import uuid from openstackclient.tests.functional import base @@ -17,33 +18,138 @@ from openstackclient.tests.functional import base class AddressScopeTests(base.TestCase): """Functional tests for address scope. """ - NAME = uuid.uuid4().hex - HEADERS = ['Name'] - FIELDS = ['name'] + + # NOTE(dtroyer): Do not normalize the setup and teardown of the resource + # creation and deletion. Little is gained when each test + # has its own needs and there are collisions when running + # tests in parallel. @classmethod def setUpClass(cls): - opts = cls.get_opts(cls.FIELDS) - raw_output = cls.openstack('address scope create ' + cls.NAME + opts) - cls.assertOutput(cls.NAME + "\n", raw_output) + # Set up some regex for matching below + cls.re_name = re.compile("name\s+\|\s+([^|]+?)\s+\|") + cls.re_ip_version = re.compile("ip_version\s+\|\s+(\S+)") + cls.re_shared = re.compile("shared\s+\|\s+(\S+)") - @classmethod - def tearDownClass(cls): - raw_output = cls.openstack('address scope delete ' + cls.NAME) - cls.assertOutput('', raw_output) + def test_address_scope_delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + raw_output = self.openstack( + 'address scope create ' + name1, + ) + self.assertEqual( + name1, + re.search(self.re_name, raw_output).group(1), + ) + # Check the default values + self.assertEqual( + 'False', + re.search(self.re_shared, raw_output).group(1), + ) + + name2 = uuid.uuid4().hex + raw_output = self.openstack( + 'address scope create ' + name2, + ) + self.assertEqual( + name2, + re.search(self.re_name, raw_output).group(1), + ) + + raw_output = self.openstack( + 'address scope delete ' + name1 + ' ' + name2, + ) + self.assertOutput('', raw_output) def test_address_scope_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('address scope list' + opts) - self.assertIn(self.NAME, raw_output) + """Test create defaults, list filters, delete""" + name1 = uuid.uuid4().hex + raw_output = self.openstack( + 'address scope create --ip-version 4 --share ' + name1, + ) + self.addCleanup(self.openstack, 'address scope delete ' + name1) + self.assertEqual( + '4', + re.search(self.re_ip_version, raw_output).group(1), + ) + self.assertEqual( + 'True', + re.search(self.re_shared, raw_output).group(1), + ) + + name2 = uuid.uuid4().hex + raw_output = self.openstack( + 'address scope create --ip-version 6 --no-share ' + name2, + ) + self.addCleanup(self.openstack, 'address scope delete ' + name2) + self.assertEqual( + '6', + re.search(self.re_ip_version, raw_output).group(1), + ) + self.assertEqual( + 'False', + re.search(self.re_shared, raw_output).group(1), + ) - def test_address_scope_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('address scope show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) + # Test list + raw_output = self.openstack('address scope list') + self.assertIsNotNone(re.search(name1 + "\s+\|\s+4", raw_output)) + self.assertIsNotNone(re.search(name2 + "\s+\|\s+6", raw_output)) + + # Test list --share + # TODO(dtroyer): returns 'HttpException: Bad Request' + # raw_output = self.openstack('address scope list --share') + # self.assertIsNotNone(re.search(name1 + "\s+\|\s+4", raw_output)) + # self.assertIsNotNone(re.search(name2 + "\s+\|\s+6", raw_output)) + + # Test list --no-share + # TODO(dtroyer): returns 'HttpException: Bad Request' + # raw_output = self.openstack('address scope list --no-share') + # self.assertIsNotNone(re.search(name1 + "\s+\|\s+4", raw_output)) + # self.assertIsNotNone(re.search(name2 + "\s+\|\s+6", raw_output)) def test_address_scope_set(self): - self.openstack('address scope set --share ' + self.NAME) - opts = self.get_opts(['shared']) - raw_output = self.openstack('address scope show ' + self.NAME + opts) - self.assertEqual("True\n", raw_output) + """Tests create options, set, show, delete""" + name = uuid.uuid4().hex + newname = name + "_" + raw_output = self.openstack( + 'address scope create ' + + '--ip-version 4 ' + + '--no-share ' + + name, + ) + self.addCleanup(self.openstack, 'address scope delete ' + newname) + self.assertEqual( + name, + re.search(self.re_name, raw_output).group(1), + ) + self.assertEqual( + '4', + re.search(self.re_ip_version, raw_output).group(1), + ) + self.assertEqual( + 'False', + re.search(self.re_shared, raw_output).group(1), + ) + + raw_output = self.openstack( + 'address scope set ' + + '--name ' + newname + + ' --share ' + + name, + ) + self.assertOutput('', raw_output) + + raw_output = self.openstack('address scope show ' + newname) + self.assertEqual( + newname, + re.search(self.re_name, raw_output).group(1), + ) + self.assertEqual( + '4', + re.search(self.re_ip_version, raw_output).group(1), + ) + self.assertEqual( + 'True', + re.search(self.re_shared, raw_output).group(1), + ) diff --git a/openstackclient/tests/functional/network/v2/test_floating_ip.py b/openstackclient/tests/functional/network/v2/test_floating_ip.py index f3a1971f..8fbec3d5 100644 --- a/openstackclient/tests/functional/network/v2/test_floating_ip.py +++ b/openstackclient/tests/functional/network/v2/test_floating_ip.py @@ -10,49 +10,161 @@ # License for the specific language governing permissions and limitations # under the License. +import random +import re import uuid from openstackclient.tests.functional import base class FloatingIpTests(base.TestCase): - """Functional tests for floating ip. """ + """Functional tests for floating ip""" SUBNET_NAME = uuid.uuid4().hex NETWORK_NAME = uuid.uuid4().hex - ID = None - HEADERS = ['ID'] - FIELDS = ['id'] @classmethod def setUpClass(cls): - # Create a network for the floating ip. - cls.openstack('network create --external ' + cls.NETWORK_NAME) - # Create a subnet for the network. - cls.openstack( - 'subnet create --network ' + cls.NETWORK_NAME + - ' --subnet-range 10.10.10.0/24 ' + - cls.SUBNET_NAME - ) - opts = cls.get_opts(cls.FIELDS) + # Set up some regex for matching below + cls.re_id = re.compile("id\s+\|\s+(\S+)") + cls.re_floating_ip = re.compile("floating_ip_address\s+\|\s+(\S+)") + cls.re_fixed_ip = re.compile("fixed_ip_address\s+\|\s+(\S+)") + cls.re_description = re.compile("description\s+\|\s+([^|]+?)\s+\|") + cls.re_network_id = re.compile("floating_network_id\s+\|\s+(\S+)") + + # Create a network for the floating ip raw_output = cls.openstack( - 'floating ip create ' + cls.NETWORK_NAME + opts) - cls.ID = raw_output.strip('\n') + 'network create --external ' + cls.NETWORK_NAME + ) + cls.network_id = re.search(cls.re_id, raw_output).group(1) + + # Try random subnet range for subnet creating + # Because we can not determine ahead of time what subnets are already + # in use, possibly by another test running in parallel, try 4 times + for i in range(4): + # Make a random subnet + cls.subnet = ".".join(map( + str, + (random.randint(0, 223) for _ in range(3)) + )) + ".0/26" + try: + # Create a subnet for the network + raw_output = cls.openstack( + 'subnet create ' + + '--network ' + cls.NETWORK_NAME + ' ' + + '--subnet-range ' + cls.subnet + ' ' + + cls.SUBNET_NAME + ) + except Exception: + if (i == 3): + # raise the exception at the last time + raise + pass + else: + # break and no longer retry if create sucessfully + break + + cls.subnet_id = re.search(cls.re_id, raw_output).group(1) @classmethod def tearDownClass(cls): - raw_output = cls.openstack('floating ip delete ' + cls.ID) - cls.assertOutput('', raw_output) raw_output = cls.openstack('subnet delete ' + cls.SUBNET_NAME) cls.assertOutput('', raw_output) raw_output = cls.openstack('network delete ' + cls.NETWORK_NAME) cls.assertOutput('', raw_output) + def test_floating_ip_delete(self): + """Test create, delete multiple""" + raw_output = self.openstack( + 'floating ip create ' + + '--description aaaa ' + + self.NETWORK_NAME + ) + re_ip = re.search(self.re_floating_ip, raw_output) + self.assertIsNotNone(re_ip) + ip1 = re_ip.group(1) + self.assertEqual( + 'aaaa', + re.search(self.re_description, raw_output).group(1), + ) + + raw_output = self.openstack( + 'floating ip create ' + + '--description bbbb ' + + self.NETWORK_NAME + ) + ip2 = re.search(self.re_floating_ip, raw_output).group(1) + self.assertEqual( + 'bbbb', + re.search(self.re_description, raw_output).group(1), + ) + + # Clean up after ourselves + raw_output = self.openstack('floating ip delete ' + ip1 + ' ' + ip2) + self.assertOutput('', raw_output) + def test_floating_ip_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('floating ip list' + opts) - self.assertIn(self.ID, raw_output) + """Test create defaults, list filters, delete""" + raw_output = self.openstack( + 'floating ip create ' + + '--description aaaa ' + + self.NETWORK_NAME + ) + re_ip = re.search(self.re_floating_ip, raw_output) + self.assertIsNotNone(re_ip) + ip1 = re_ip.group(1) + self.addCleanup(self.openstack, 'floating ip delete ' + ip1) + self.assertEqual( + 'aaaa', + re.search(self.re_description, raw_output).group(1), + ) + self.assertIsNotNone(re.search(self.re_network_id, raw_output)) + + raw_output = self.openstack( + 'floating ip create ' + + '--description bbbb ' + + self.NETWORK_NAME + ) + ip2 = re.search(self.re_floating_ip, raw_output).group(1) + self.addCleanup(self.openstack, 'floating ip delete ' + ip2) + self.assertEqual( + 'bbbb', + re.search(self.re_description, raw_output).group(1), + ) + + # Test list + raw_output = self.openstack('floating ip list') + self.assertIsNotNone(re.search("\|\s+" + ip1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + ip2 + "\s+\|", raw_output)) + + # Test list --long + raw_output = self.openstack('floating ip list --long') + self.assertIsNotNone(re.search("\|\s+" + ip1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + ip2 + "\s+\|", raw_output)) + + # TODO(dtroyer): add more filter tests def test_floating_ip_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('floating ip show ' + self.ID + opts) - self.assertEqual(self.ID + "\n", raw_output) + """Test show""" + raw_output = self.openstack( + 'floating ip create ' + + '--description shosho ' + + # '--fixed-ip-address 1.2.3.4 ' + + self.NETWORK_NAME + ) + re_ip = re.search(self.re_floating_ip, raw_output) + self.assertIsNotNone(re_ip) + ip = re_ip.group(1) + + raw_output = self.openstack('floating ip show ' + ip) + self.addCleanup(self.openstack, 'floating ip delete ' + ip) + + self.assertEqual( + 'shosho', + re.search(self.re_description, raw_output).group(1), + ) + # TODO(dtroyer): not working??? + # self.assertEqual( + # '1.2.3.4', + # re.search(self.re_floating_ip, raw_output).group(1), + # ) + self.assertIsNotNone(re.search(self.re_network_id, raw_output)) diff --git a/openstackclient/tests/functional/network/v2/test_meter.py b/openstackclient/tests/functional/network/v2/test_meter.py new file mode 100644 index 00000000..7dce34e7 --- /dev/null +++ b/openstackclient/tests/functional/network/v2/test_meter.py @@ -0,0 +1,102 @@ +# Copyright (c) 2016, Intel Corporation. +# 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 re +import uuid + +from openstackclient.tests.functional import base + + +class TestMeter(base.TestCase): + """Functional tests for network meter.""" + + # NOTE(dtroyer): Do not normalize the setup and teardown of the resource + # creation and deletion. Little is gained when each test + # has its own needs and there are collisions when running + # tests in parallel. + + @classmethod + def setUpClass(cls): + # Set up some regex for matching below + cls.re_name = re.compile("name\s+\|\s+([^|]+?)\s+\|") + cls.re_shared = re.compile("shared\s+\|\s+(\S+)") + cls.re_description = re.compile("description\s+\|\s+([^|]+?)\s+\|") + + def test_meter_delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + name2 = uuid.uuid4().hex + + raw_output = self.openstack( + 'network meter create ' + name1, + ) + self.assertEqual( + name1, + re.search(self.re_name, raw_output).group(1), + ) + # Check if default shared values + self.assertEqual( + 'False', + re.search(self.re_shared, raw_output).group(1) + ) + + raw_output = self.openstack( + 'network meter create ' + name2, + ) + self.assertEqual( + name2, + re.search(self.re_name, raw_output).group(1), + ) + + raw_output = self.openstack( + 'network meter delete ' + name1 + ' ' + name2, + ) + self.assertOutput('', raw_output) + + def test_meter_list(self): + """Test create, list filters, delete""" + name1 = uuid.uuid4().hex + raw_output = self.openstack( + 'network meter create --description Test1 --share ' + name1, + ) + self.addCleanup(self.openstack, 'network meter delete ' + name1) + + self.assertEqual( + 'Test1', + re.search(self.re_description, raw_output).group(1), + ) + self.assertEqual( + 'True', + re.search(self.re_shared, raw_output).group(1), + ) + + name2 = uuid.uuid4().hex + raw_output = self.openstack( + 'network meter create --description Test2 --no-share ' + name2, + ) + self.addCleanup(self.openstack, 'network meter delete ' + name2) + + self.assertEqual( + 'Test2', + re.search(self.re_description, raw_output).group(1), + ) + self.assertEqual( + 'False', + re.search(self.re_shared, raw_output).group(1), + ) + + raw_output = self.openstack('network meter list') + self.assertIsNotNone(re.search(name1 + "\s+\|\s+Test1", raw_output)) + self.assertIsNotNone(re.search(name2 + "\s+\|\s+Test2", raw_output)) diff --git a/openstackclient/tests/functional/network/v2/test_network.py b/openstackclient/tests/functional/network/v2/test_network.py index c77ff642..c55d70f9 100644 --- a/openstackclient/tests/functional/network/v2/test_network.py +++ b/openstackclient/tests/functional/network/v2/test_network.py @@ -10,41 +10,277 @@ # License for the specific language governing permissions and limitations # under the License. +import json import uuid from openstackclient.tests.functional import base class NetworkTests(base.TestCase): - """Functional tests for network. """ - NAME = uuid.uuid4().hex - HEADERS = ['Name'] - FIELDS = ['name'] - - @classmethod - def setUpClass(cls): - opts = cls.get_opts(cls.FIELDS) - raw_output = cls.openstack('network create ' + cls.NAME + opts) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) - - @classmethod - def tearDownClass(cls): - raw_output = cls.openstack('network delete ' + cls.NAME) - cls.assertOutput('', raw_output) + """Functional tests for network""" + + def test_network_delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'network create -f json ' + + '--description aaaa ' + + name1 + )) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'network create -f json ' + + '--description bbbb ' + + name2 + )) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + + del_output = self.openstack('network delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) def test_network_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('network list' + opts) - self.assertIn(self.NAME, raw_output) + """Test create defaults, list filters, delete""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'network create -f json ' + + '--description aaaa ' + + name1 + )) + self.addCleanup(self.openstack, 'network delete ' + name1) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + # Check the default values + self.assertEqual( + 'UP', + cmd_output["admin_state_up"], + ) + self.assertEqual( + False, + cmd_output["shared"], + ) + self.assertEqual( + 'Internal', + cmd_output["router:external"], + ) + # NOTE(dtroyer): is_default is not present in the create output + # so make sure it stays that way. + # NOTE(stevemar): is_default *is* present in SDK 0.9.11 and newer, + # but the value seems to always be None, regardless + # of the --default or --no-default value. + # self.assertEqual('x', cmd_output) + if ('is_default' in cmd_output): + self.assertEqual( + None, + cmd_output["is_default"], + ) + self.assertEqual( + True, + cmd_output["port_security_enabled"], + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'network create -f json ' + + '--description bbbb ' + + '--disable ' + + '--share ' + + name2 + )) + self.addCleanup(self.openstack, 'network delete ' + name2) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + self.assertEqual( + 'DOWN', + cmd_output["admin_state_up"], + ) + self.assertEqual( + True, + cmd_output["shared"], + ) + if ('is_default' in cmd_output): + self.assertEqual( + None, + cmd_output["is_default"], + ) + self.assertEqual( + True, + cmd_output["port_security_enabled"], + ) + + # Test list --long + cmd_output = json.loads(self.openstack( + "network list -f json " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertIn(name2, col_name) + + # Test list --long --enable + cmd_output = json.loads(self.openstack( + "network list -f json " + + "--enable " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertNotIn(name2, col_name) + + # Test list --long --disable + cmd_output = json.loads(self.openstack( + "network list -f json " + + "--disable " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, col_name) + self.assertIn(name2, col_name) + + # Test list --long --share + cmd_output = json.loads(self.openstack( + "network list -f json " + + "--share " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, col_name) + self.assertIn(name2, col_name) + + # Test list --long --no-share + cmd_output = json.loads(self.openstack( + "network list -f json " + + "--no-share " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertNotIn(name2, col_name) def test_network_set(self): - raw_output = self.openstack('network set --disable ' + self.NAME) - opts = self.get_opts(['name', 'admin_state_up']) - raw_output = self.openstack('network show ' + self.NAME + opts) - self.assertEqual("DOWN\n" + self.NAME + "\n", raw_output) - - def test_network_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('network show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) + """Tests create options, set, show, delete""" + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'network create -f json ' + + '--description aaaa ' + + '--enable ' + + '--no-share ' + + '--internal ' + + '--no-default ' + + '--enable-port-security ' + + name + )) + self.addCleanup(self.openstack, 'network delete ' + name) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + self.assertEqual( + 'UP', + cmd_output["admin_state_up"], + ) + self.assertEqual( + False, + cmd_output["shared"], + ) + self.assertEqual( + 'Internal', + cmd_output["router:external"], + ) + # NOTE(dtroyer): is_default is not present in the create output + # so make sure it stays that way. + # NOTE(stevemar): is_default *is* present in SDK 0.9.11 and newer, + # but the value seems to always be None, regardless + # of the --default or --no-default value. + if ('is_default' in cmd_output): + self.assertEqual( + None, + cmd_output["is_default"], + ) + self.assertEqual( + True, + cmd_output["port_security_enabled"], + ) + + raw_output = self.openstack( + 'network set ' + + '--description cccc ' + + '--disable ' + + '--share ' + + '--external ' + + '--disable-port-security ' + + name + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'network show -f json ' + name + )) + + self.assertEqual( + 'cccc', + cmd_output["description"], + ) + self.assertEqual( + 'DOWN', + cmd_output["admin_state_up"], + ) + self.assertEqual( + True, + cmd_output["shared"], + ) + self.assertEqual( + 'External', + cmd_output["router:external"], + ) + # why not 'None' like above?? + self.assertEqual( + False, + cmd_output["is_default"], + ) + self.assertEqual( + False, + cmd_output["port_security_enabled"], + ) + + # NOTE(dtroyer): There is ambiguity around is_default in that + # it is not in the API docs and apparently can + # not be set when the network is --external, + # although the option handling code only looks at + # the value of is_default when external is True. + raw_output = self.openstack( + 'network set ' + + '--default ' + + name + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'network show -f json ' + name + )) + + self.assertEqual( + 'cccc', + cmd_output["description"], + ) + # NOTE(dtroyer): This should be 'True' + self.assertEqual( + False, + cmd_output["port_security_enabled"], + ) diff --git a/openstackclient/tests/functional/network/v2/test_network_qos_rule.py b/openstackclient/tests/functional/network/v2/test_network_qos_rule.py new file mode 100644 index 00000000..af0c9bac --- /dev/null +++ b/openstackclient/tests/functional/network/v2/test_network_qos_rule.py @@ -0,0 +1,181 @@ +# Copyright (c) 2016, Intel Corporation. +# 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 uuid + +from openstackclient.tests.functional import base + + +class NetworkQosRuleTestsMinimumBandwidth(base.TestCase): + """Functional tests for QoS minimum bandwidth rule.""" + RULE_ID = None + QOS_POLICY_NAME = 'qos_policy_' + uuid.uuid4().hex + MIN_KBPS = 2800 + MIN_KBPS_MODIFIED = 7500 + DIRECTION = '--egress' + HEADERS = ['ID'] + FIELDS = ['id'] + TYPE = 'minimum-bandwidth' + + @classmethod + def setUpClass(cls): + opts = cls.get_opts(cls.FIELDS) + cls.openstack('network qos policy create ' + cls.QOS_POLICY_NAME) + cls.RULE_ID = cls.openstack('network qos rule create --type ' + + cls.TYPE + ' --min-kbps ' + + str(cls.MIN_KBPS) + ' ' + cls.DIRECTION + + ' ' + cls.QOS_POLICY_NAME + opts) + cls.assertsOutputNotNone(cls.RULE_ID) + + @classmethod + def tearDownClass(cls): + raw_output = cls.openstack('network qos rule delete ' + + cls.QOS_POLICY_NAME + ' ' + cls.RULE_ID) + cls.openstack('network qos policy delete ' + cls.QOS_POLICY_NAME) + cls.assertOutput('', raw_output) + + def test_qos_policy_list(self): + opts = self.get_opts(self.HEADERS) + raw_output = self.openstack('network qos rule list ' + + self.QOS_POLICY_NAME + opts) + self.assertIn(self.RULE_ID, raw_output) + + def test_qos_policy_show(self): + opts = self.get_opts(self.FIELDS) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(self.RULE_ID, raw_output) + + def test_qos_policy_set(self): + self.openstack('network qos rule set --min-kbps ' + + str(self.MIN_KBPS_MODIFIED) + ' ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID) + opts = self.get_opts(['min_kbps']) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(str(self.MIN_KBPS_MODIFIED) + "\n", raw_output) + + +class NetworkQosRuleTestsDSCPMarking(base.TestCase): + """Functional tests for QoS DSCP marking rule.""" + RULE_ID = None + QOS_POLICY_NAME = 'qos_policy_' + uuid.uuid4().hex + DSCP_MARK = 8 + DSCP_MARK_MODIFIED = 32 + HEADERS = ['ID'] + FIELDS = ['id'] + TYPE = 'dscp-marking' + + @classmethod + def setUpClass(cls): + opts = cls.get_opts(cls.FIELDS) + cls.openstack('network qos policy create ' + cls.QOS_POLICY_NAME) + cls.RULE_ID = cls.openstack('network qos rule create --type ' + + cls.TYPE + ' --dscp-mark ' + + str(cls.DSCP_MARK) + ' ' + + cls.QOS_POLICY_NAME + opts) + cls.assertsOutputNotNone(cls.RULE_ID) + + @classmethod + def tearDownClass(cls): + raw_output = cls.openstack('network qos rule delete ' + + cls.QOS_POLICY_NAME + ' ' + cls.RULE_ID) + cls.openstack('network qos policy delete ' + cls.QOS_POLICY_NAME) + cls.assertOutput('', raw_output) + + def test_qos_policy_list(self): + opts = self.get_opts(self.HEADERS) + raw_output = self.openstack('network qos rule list ' + + self.QOS_POLICY_NAME + opts) + self.assertIn(self.RULE_ID, raw_output) + + def test_qos_policy_show(self): + opts = self.get_opts(self.FIELDS) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(self.RULE_ID, raw_output) + + def test_qos_policy_set(self): + self.openstack('network qos rule set --dscp-mark ' + + str(self.DSCP_MARK_MODIFIED) + ' ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID) + opts = self.get_opts(['dscp_mark']) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(str(self.DSCP_MARK_MODIFIED) + "\n", raw_output) + + +class NetworkQosRuleTestsBandwidthLimit(base.TestCase): + """Functional tests for QoS bandwidth limit rule.""" + RULE_ID = None + QOS_POLICY_NAME = 'qos_policy_' + uuid.uuid4().hex + MAX_KBPS = 10000 + MAX_KBPS_MODIFIED = 15000 + MAX_BURST_KBITS = 1400 + MAX_BURST_KBITS_MODIFIED = 1800 + HEADERS = ['ID'] + FIELDS = ['id'] + TYPE = 'bandwidth-limit' + + @classmethod + def setUpClass(cls): + opts = cls.get_opts(cls.FIELDS) + cls.openstack('network qos policy create ' + cls.QOS_POLICY_NAME) + cls.RULE_ID = cls.openstack('network qos rule create --type ' + + cls.TYPE + ' --max-kbps ' + + str(cls.MAX_KBPS) + ' --max-burst-kbits ' + + str(cls.MAX_BURST_KBITS) + ' ' + + cls.QOS_POLICY_NAME + opts) + cls.assertsOutputNotNone(cls.RULE_ID) + + @classmethod + def tearDownClass(cls): + raw_output = cls.openstack('network qos rule delete ' + + cls.QOS_POLICY_NAME + ' ' + cls.RULE_ID) + cls.openstack('network qos policy delete ' + cls.QOS_POLICY_NAME) + cls.assertOutput('', raw_output) + + def test_qos_policy_list(self): + opts = self.get_opts(self.HEADERS) + raw_output = self.openstack('network qos rule list ' + + self.QOS_POLICY_NAME + opts) + self.assertIn(self.RULE_ID, raw_output) + + def test_qos_policy_show(self): + opts = self.get_opts(self.FIELDS) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(self.RULE_ID, raw_output) + + def test_qos_policy_set(self): + self.openstack('network qos rule set --max-kbps ' + + str(self.MAX_KBPS_MODIFIED) + ' --max-burst-kbits ' + + str(self.MAX_BURST_KBITS_MODIFIED) + ' ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID) + opts = self.get_opts(['max_kbps']) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(str(self.MAX_KBPS_MODIFIED) + "\n", raw_output) + opts = self.get_opts(['max_burst_kbps']) + raw_output = self.openstack('network qos rule show ' + + self.QOS_POLICY_NAME + ' ' + self.RULE_ID + + opts) + self.assertEqual(str(self.MAX_BURST_KBITS_MODIFIED) + "\n", raw_output) diff --git a/openstackclient/tests/functional/network/v2/test_network_qos_rule_type.py b/openstackclient/tests/functional/network/v2/test_network_qos_rule_type.py new file mode 100644 index 00000000..7dff0cbd --- /dev/null +++ b/openstackclient/tests/functional/network/v2/test_network_qos_rule_type.py @@ -0,0 +1,28 @@ +# Copyright (c) 2016, Intel Corporation. +# 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. + +from openstackclient.tests.functional import base + + +class NetworkQosRuleTypeTests(base.TestCase): + """Functional tests for Network QoS rule type. """ + + AVAILABLE_RULE_TYPES = ['dscp_marking', + 'bandwidth_limit'] + + def test_qos_rule_type_list(self): + raw_output = self.openstack('network qos rule type list') + for rule_type in self.AVAILABLE_RULE_TYPES: + self.assertIn(rule_type, raw_output) diff --git a/openstackclient/tests/functional/network/v2/test_network_service_provider.py b/openstackclient/tests/functional/network/v2/test_network_service_provider.py index 379de430..6fbff6c8 100644 --- a/openstackclient/tests/functional/network/v2/test_network_service_provider.py +++ b/openstackclient/tests/functional/network/v2/test_network_service_provider.py @@ -19,7 +19,7 @@ from openstackclient.tests.functional import base class TestNetworkServiceProvider(base.TestCase): """Functional tests for network service provider""" - SERVICE_TYPE = ['L3_ROUTER_NAT'] + SERVICE_TYPE = 'L3_ROUTER_NAT' def test_network_service_provider_list(self): raw_output = self.openstack('network service provider list') diff --git a/openstackclient/tests/functional/network/v2/test_port.py b/openstackclient/tests/functional/network/v2/test_port.py index decd9553..e100bd82 100644 --- a/openstackclient/tests/functional/network/v2/test_port.py +++ b/openstackclient/tests/functional/network/v2/test_port.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import re import uuid from openstackclient.tests.functional import base @@ -24,35 +25,152 @@ class PortTests(base.TestCase): @classmethod def setUpClass(cls): - # Create a network for the subnet. - cls.openstack('network create ' + cls.NETWORK_NAME) - opts = cls.get_opts(cls.FIELDS) - raw_output = cls.openstack( - 'port create --network ' + cls.NETWORK_NAME + ' ' + - cls.NAME + opts - ) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) + # Set up some regex for matching below + cls.re_id = re.compile("\s+id\s+\|\s+(\S+)") + cls.re_name = re.compile("\s+name\s+\|\s+([^|]+?)\s+\|") + cls.re_description = re.compile("\s+description\s+\|\s+([^|]+?)\s+\|") + cls.re_mac_address = re.compile("\s+mac_address\s+\|\s+([^|]+?)\s+\|") + cls.re_state = re.compile("\s+admin_state_up\s+\|\s+([^|]+?)\s+\|") + + # Create a network for the port + raw_output = cls.openstack('network create ' + cls.NETWORK_NAME) + cls.network_id = re.search(cls.re_id, raw_output).group(1) @classmethod def tearDownClass(cls): - raw_output = cls.openstack('port delete ' + cls.NAME) - cls.assertOutput('', raw_output) raw_output = cls.openstack('network delete ' + cls.NETWORK_NAME) cls.assertOutput('', raw_output) + def test_port_delete(self): + """Test create, delete multiple""" + raw_output = self.openstack( + 'port create --network ' + self.NETWORK_NAME + ' ' + self.NAME + ) + re_id1 = re.search(self.re_id, raw_output) + self.assertIsNotNone(re_id1) + id1 = re_id1.group(1) + self.assertIsNotNone( + re.search(self.re_mac_address, raw_output).group(1), + ) + self.assertEqual( + self.NAME, + re.search(self.re_name, raw_output).group(1), + ) + + raw_output = self.openstack( + 'port create ' + + '--network ' + self.NETWORK_NAME + ' ' + + self.NAME + 'x' + ) + id2 = re.search(self.re_id, raw_output).group(1) + self.assertIsNotNone( + re.search(self.re_mac_address, raw_output).group(1), + ) + self.assertEqual( + self.NAME + 'x', + re.search(self.re_name, raw_output).group(1), + ) + + # Clean up after ourselves + raw_output = self.openstack('port delete ' + id1 + ' ' + id2) + self.assertOutput('', raw_output) + def test_port_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('port list' + opts) - self.assertIn(self.NAME, raw_output) + """Test create defaults, list, delete""" + raw_output = self.openstack( + 'port create --network ' + self.NETWORK_NAME + ' ' + self.NAME + ) + re_id1 = re.search(self.re_id, raw_output) + self.assertIsNotNone(re_id1) + id1 = re_id1.group(1) + mac1 = re.search(self.re_mac_address, raw_output).group(1) + self.addCleanup(self.openstack, 'port delete ' + id1) + self.assertEqual( + self.NAME, + re.search(self.re_name, raw_output).group(1), + ) + + raw_output = self.openstack( + 'port create ' + + '--network ' + self.NETWORK_NAME + ' ' + + self.NAME + 'x' + ) + id2 = re.search(self.re_id, raw_output).group(1) + mac2 = re.search(self.re_mac_address, raw_output).group(1) + self.addCleanup(self.openstack, 'port delete ' + id2) + self.assertEqual( + self.NAME + 'x', + re.search(self.re_name, raw_output).group(1), + ) + + # Test list + raw_output = self.openstack('port list') + self.assertIsNotNone(re.search("\|\s+" + id1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + id2 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + mac1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + mac2 + "\s+\|", raw_output)) + + # Test list --long + raw_output = self.openstack('port list --long') + self.assertIsNotNone(re.search("\|\s+" + id1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + id2 + "\s+\|", raw_output)) + + # Test list --mac-address + raw_output = self.openstack('port list --mac-address ' + mac2) + self.assertIsNone(re.search("\|\s+" + id1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + id2 + "\s+\|", raw_output)) + self.assertIsNone(re.search("\|\s+" + mac1 + "\s+\|", raw_output)) + self.assertIsNotNone(re.search("\|\s+" + mac2 + "\s+\|", raw_output)) def test_port_set(self): - self.openstack('port set --disable ' + self.NAME) - opts = self.get_opts(['name', 'admin_state_up']) - raw_output = self.openstack('port show ' + self.NAME + opts) - self.assertEqual("DOWN\n" + self.NAME + "\n", raw_output) - - def test_port_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('port show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) + """Test create, set, show, delete""" + raw_output = self.openstack( + 'port create ' + + '--network ' + self.NETWORK_NAME + ' ' + + '--description xyzpdq ' + '--disable ' + + self.NAME + ) + re_id = re.search(self.re_id, raw_output) + self.assertIsNotNone(re_id) + id = re_id.group(1) + self.addCleanup(self.openstack, 'port delete ' + id) + self.assertEqual( + self.NAME, + re.search(self.re_name, raw_output).group(1), + ) + self.assertEqual( + 'xyzpdq', + re.search(self.re_description, raw_output).group(1), + ) + self.assertEqual( + 'DOWN', + re.search(self.re_state, raw_output).group(1), + ) + + raw_output = self.openstack( + 'port set ' + + '--enable ' + + self.NAME + ) + self.assertOutput('', raw_output) + + raw_output = self.openstack( + 'port show ' + + self.NAME + ) + self.assertEqual( + self.NAME, + re.search(self.re_name, raw_output).group(1), + ) + self.assertEqual( + 'xyzpdq', + re.search(self.re_description, raw_output).group(1), + ) + self.assertEqual( + 'UP', + re.search(self.re_state, raw_output).group(1), + ) + self.assertIsNotNone( + re.search(self.re_mac_address, raw_output).group(1), + ) |