diff options
Diffstat (limited to 'openstackclient/tests')
55 files changed, 4647 insertions, 484 deletions
diff --git a/openstackclient/tests/functional/base.py b/openstackclient/tests/functional/base.py index 885abc02..85743296 100644 --- a/openstackclient/tests/functional/base.py +++ b/openstackclient/tests/functional/base.py @@ -16,7 +16,6 @@ import shlex import subprocess import testtools -import six from tempest.lib.cli import output_parser from tempest.lib import exceptions @@ -77,29 +76,28 @@ class TestCase(testtools.TestCase): if expected not in actual: raise Exception(expected + ' not in ' + actual) + @classmethod + def assertsOutputNotNone(cls, observed): + if observed is None: + raise Exception('No output observed') + def assert_table_structure(self, items, field_names): """Verify that all items have keys listed in field_names.""" for item in items: for field in field_names: self.assertIn(field, item) - def assert_show_fields(self, items, field_names): + def assert_show_fields(self, show_output, field_names): """Verify that all items have keys listed in field_names.""" - for item in items: - for key in six.iterkeys(item): - self.assertIn(key, field_names) - - def assert_show_structure(self, items, field_names): - """Verify that all field_names listed in keys of all items.""" - if isinstance(items, list): - o = {} - for d in items: - o.update(d) - else: - o = items - item_keys = o.keys() - for field in field_names: - self.assertIn(field, item_keys) + + # field_names = ['name', 'description'] + # show_output = [{'name': 'fc2b98d8faed4126b9e371eda045ade2'}, + # {'description': 'description-821397086'}] + # this next line creates a flattened list of all 'keys' (like 'name', + # and 'description' out of the output + all_headers = [item for sublist in show_output for item in sublist] + for field_name in field_names: + self.assertIn(field_name, all_headers) def parse_show_as_object(self, raw_output): """Return a dict with values parsed from cli output.""" diff --git a/openstackclient/tests/functional/common/test_help.py b/openstackclient/tests/functional/common/test_help.py index bbc52197..211c52b1 100644 --- a/openstackclient/tests/functional/common/test_help.py +++ b/openstackclient/tests/functional/common/test_help.py @@ -64,3 +64,10 @@ class HelpTests(base.TestCase): raw_output = self.openstack('help server') for command in [row[0] for row in self.SERVER_COMMANDS]: self.assertIn(command, raw_output) + + def test_networking_commands_help(self): + """Check networking related commands in help message.""" + raw_output = self.openstack('help network list') + self.assertIn('List networks', raw_output) + raw_output = self.openstack('network create --help') + self.assertIn('Create new network', raw_output) diff --git a/openstackclient/tests/functional/compute/v2/test_aggregate.py b/openstackclient/tests/functional/compute/v2/test_aggregate.py index 2bc88e7b..38368103 100644 --- a/openstackclient/tests/functional/compute/v2/test_aggregate.py +++ b/openstackclient/tests/functional/compute/v2/test_aggregate.py @@ -48,7 +48,7 @@ class AggregateTests(base.TestCase): self.assertEqual(self.NAME + "\n", raw_output) def test_aggregate_properties(self): - opts = self.get_opts(['properties']) + opts = self.get_opts(['name', 'properties']) raw_output = self.openstack( 'aggregate set --property a=b --property c=d ' + self.NAME @@ -56,7 +56,7 @@ class AggregateTests(base.TestCase): self.assertEqual('', raw_output) raw_output = self.openstack('aggregate show ' + self.NAME + opts) - self.assertIn("a='b', c='d'\n", raw_output) + self.assertIn(self.NAME + "\na='b', c='d'\n", raw_output) raw_output = self.openstack( 'aggregate unset --property a ' + self.NAME @@ -64,4 +64,44 @@ class AggregateTests(base.TestCase): self.assertEqual('', raw_output) raw_output = self.openstack('aggregate show ' + self.NAME + opts) - self.assertIn("c='d'\n", raw_output) + self.assertIn(self.NAME + "\nc='d'\n", raw_output) + + raw_output = self.openstack( + 'aggregate set --property a=b --property c=d ' + self.NAME + ) + self.assertEqual('', raw_output) + + raw_output = self.openstack( + 'aggregate set --no-property ' + self.NAME + ) + self.assertEqual('', raw_output) + + raw_output = self.openstack('aggregate show ' + self.NAME + opts) + self.assertNotIn("a='b', c='d'", raw_output) + + def test_aggregate_set(self): + opts = self.get_opts(["name", "availability_zone"]) + + raw_output = self.openstack( + 'aggregate set --zone Zone_1 ' + self.NAME) + self.assertEqual("", raw_output) + + raw_output = self.openstack('aggregate show ' + self.NAME + opts) + self.assertEqual("Zone_1\n" + self.NAME + "\n", raw_output) + + def test_aggregate_add_and_remove_host(self): + opts = self.get_opts(["hosts", "name"]) + + raw_output = self.openstack('host list -f value -c "Host Name"') + host_name = raw_output.split()[0] + + self.openstack( + 'aggregate add host ' + self.NAME + ' ' + host_name) + raw_output = self.openstack('aggregate show ' + self.NAME + opts) + self.assertEqual("[u'" + host_name + "']" + "\n" + self.NAME + "\n", + raw_output) + + self.openstack( + 'aggregate remove host ' + self.NAME + ' ' + host_name) + raw_output = self.openstack('aggregate show ' + self.NAME + opts) + self.assertEqual("[]\n" + self.NAME + "\n", raw_output) diff --git a/openstackclient/tests/functional/compute/v2/test_flavor.py b/openstackclient/tests/functional/compute/v2/test_flavor.py index 794a6cc3..0b01da51 100644 --- a/openstackclient/tests/functional/compute/v2/test_flavor.py +++ b/openstackclient/tests/functional/compute/v2/test_flavor.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import json import uuid from openstackclient.tests.functional import base @@ -18,52 +19,224 @@ from openstackclient.tests.functional import base class FlavorTests(base.TestCase): """Functional tests for flavor.""" - NAME = uuid.uuid4().hex - HEADERS = ['Name'] - FIELDS = ['name'] + PROJECT_NAME = uuid.uuid4().hex @classmethod def setUpClass(cls): - opts = cls.get_opts(cls.FIELDS) - raw_output = cls.openstack( - 'flavor create --property a=b --property c=d ' + cls.NAME + opts) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) + # Make a project + cmd_output = json.loads(cls.openstack( + "project create -f json --enable " + cls.PROJECT_NAME + )) + cls.project_id = cmd_output["id"] @classmethod def tearDownClass(cls): - raw_output = cls.openstack('flavor delete ' + cls.NAME) + raw_output = cls.openstack("project delete " + cls.PROJECT_NAME) cls.assertOutput('', raw_output) + def test_flavor_delete(self): + """Test create w/project, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + "flavor create -f json " + + "--project " + self.PROJECT_NAME + " " + + "--private " + + name1 + )) + self.assertIsNotNone(cmd_output["id"]) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + "flavor create -f json " + + "--id qaz " + + "--project " + self.PROJECT_NAME + " " + + "--private " + + name2 + )) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + "qaz", + cmd_output["id"], + ) + + raw_output = self.openstack( + "flavor delete " + name1 + " " + name2, + ) + self.assertOutput('', raw_output) + def test_flavor_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('flavor list' + opts) - self.assertIn("small", raw_output) - self.assertIn(self.NAME, raw_output) + """Test create defaults, list filters, delete""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + "flavor create -f json " + + "--property a=b " + + "--property c=d " + + name1 + )) + self.addCleanup(self.openstack, "flavor delete " + name1) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + name1, + cmd_output["name"], + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + "flavor create -f json " + + "--id qaz " + + "--ram 123 " + + "--private " + + "--property a=b2 " + + "--property b=d2 " + + name2 + )) + self.addCleanup(self.openstack, "flavor delete " + name2) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + "qaz", + cmd_output["id"], + ) + self.assertEqual( + name2, + cmd_output["name"], + ) + self.assertEqual( + 123, + cmd_output["ram"], + ) + self.assertEqual( + 0, + cmd_output["disk"], + ) + self.assertEqual( + False, + cmd_output["os-flavor-access:is_public"], + ) + self.assertEqual( + "a='b2', b='d2'", + cmd_output["properties"], + ) - def test_flavor_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('flavor show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) + # Test list + cmd_output = json.loads(self.openstack( + "flavor list -f json" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertNotIn(name2, col_name) + + # Test list --long + cmd_output = json.loads(self.openstack( + "flavor list -f json " + + "--long" + )) + col_name = [x["Name"] for x in cmd_output] + col_properties = [x['Properties'] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertIn("a='b', c='d'", col_properties) + self.assertNotIn(name2, col_name) + self.assertNotIn("b2', b='d2'", col_properties) + + # Test list --public + cmd_output = json.loads(self.openstack( + "flavor list -f json " + + "--public" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertNotIn(name2, col_name) + + # Test list --private + cmd_output = json.loads(self.openstack( + "flavor list -f json " + + "--private" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, col_name) + self.assertIn(name2, col_name) + + # Test list --all + cmd_output = json.loads(self.openstack( + "flavor list -f json " + + "--all" + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + self.assertIn(name2, col_name) def test_flavor_properties(self): - opts = self.get_opts(['properties']) - # check the properties we added in create command. - raw_output = self.openstack('flavor show ' + self.NAME + opts) - self.assertEqual("a='b', c='d'\n", raw_output) + """Test create defaults, list filters, delete""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + "flavor create -f json " + + "--id qaz " + + "--ram 123 " + + "--disk 20 " + + "--private " + + "--property a=first " + + "--property b=second " + + name1 + )) + self.addCleanup(self.openstack, "flavor delete " + name1) + self.assertIsNotNone(cmd_output["id"]) + self.assertEqual( + "qaz", + cmd_output["id"], + ) + self.assertEqual( + name1, + cmd_output["name"], + ) + self.assertEqual( + 123, + cmd_output["ram"], + ) + self.assertEqual( + 20, + cmd_output["disk"], + ) + self.assertEqual( + False, + cmd_output["os-flavor-access:is_public"], + ) + self.assertEqual( + "a='first', b='second'", + cmd_output["properties"], + ) raw_output = self.openstack( - 'flavor set --property e=f --property g=h ' + self.NAME + "flavor set " + + "--property a='third and 10' " + + "--property g=fourth " + + name1 ) self.assertEqual('', raw_output) - raw_output = self.openstack('flavor show ' + self.NAME + opts) - self.assertEqual("a='b', c='d', e='f', g='h'\n", raw_output) + cmd_output = json.loads(self.openstack( + "flavor show -f json " + + name1 + )) + self.assertEqual( + "qaz", + cmd_output["id"], + ) + self.assertEqual( + "a='third and 10', b='second', g='fourth'", + cmd_output['properties'], + ) raw_output = self.openstack( - 'flavor unset --property a --property c ' + self.NAME + "flavor unset " + + "--property b " + + name1 ) self.assertEqual('', raw_output) - raw_output = self.openstack('flavor show ' + self.NAME + opts) - self.assertEqual("e='f', g='h'\n", raw_output) + cmd_output = json.loads(self.openstack( + "flavor show -f json " + + name1 + )) + self.assertEqual( + "a='third and 10', g='fourth'", + cmd_output["properties"], + ) diff --git a/openstackclient/tests/functional/identity/v2/common.py b/openstackclient/tests/functional/identity/v2/common.py index b390c5bc..ad02f779 100644 --- a/openstackclient/tests/functional/identity/v2/common.py +++ b/openstackclient/tests/functional/identity/v2/common.py @@ -22,14 +22,13 @@ BASIC_LIST_HEADERS = ['ID', 'Name'] class IdentityTests(base.TestCase): """Functional tests for Identity commands. """ - USER_FIELDS = ['email', 'enabled', 'id', 'name', 'project_id', - 'username', 'domain_id', 'default_project_id'] - PROJECT_FIELDS = ['enabled', 'id', 'name', 'description', 'domain_id'] + USER_FIELDS = ['email', 'enabled', 'id', 'name', 'project_id', 'username'] + PROJECT_FIELDS = ['enabled', 'id', 'name', 'description'] TOKEN_FIELDS = ['expires', 'id', 'project_id', 'user_id'] - ROLE_FIELDS = ['id', 'name', 'links', 'domain_id'] + ROLE_FIELDS = ['id', 'name', 'domain_id'] SERVICE_FIELDS = ['id', 'enabled', 'name', 'type', 'description'] ENDPOINT_FIELDS = ['id', 'region', 'service_id', 'service_name', - 'service_type', 'enabled', 'publicurl', + 'service_type', 'publicurl', 'adminurl', 'internalurl'] EC2_CREDENTIALS_FIELDS = ['access', 'project_id', 'secret', diff --git a/openstackclient/tests/functional/identity/v3/common.py b/openstackclient/tests/functional/identity/v3/common.py index 5dd42e70..a509574c 100644 --- a/openstackclient/tests/functional/identity/v3/common.py +++ b/openstackclient/tests/functional/identity/v3/common.py @@ -23,15 +23,15 @@ BASIC_LIST_HEADERS = ['ID', 'Name'] class IdentityTests(base.TestCase): """Functional tests for Identity commands. """ - DOMAIN_FIELDS = ['description', 'enabled', 'id', 'name', 'links'] - GROUP_FIELDS = ['description', 'domain_id', 'id', 'name', 'links'] + DOMAIN_FIELDS = ['description', 'enabled', 'id', 'name'] + GROUP_FIELDS = ['description', 'domain_id', 'id', 'name'] TOKEN_FIELDS = ['expires', 'id', 'project_id', 'user_id'] USER_FIELDS = ['email', 'enabled', 'id', 'name', 'name', 'domain_id', 'default_project_id', 'description', 'password_expires_at'] PROJECT_FIELDS = ['description', 'id', 'domain_id', 'is_domain', - 'enabled', 'name', 'parent_id', 'links'] - ROLE_FIELDS = ['id', 'name', 'links', 'domain_id'] + 'enabled', 'name', 'parent_id'] + ROLE_FIELDS = ['id', 'name', 'domain_id'] SERVICE_FIELDS = ['id', 'enabled', 'name', 'type', 'description'] REGION_FIELDS = ['description', 'enabled', 'parent_region', 'region'] ENDPOINT_FIELDS = ['id', 'region', 'region_id', 'service_id', @@ -42,7 +42,8 @@ class IdentityTests(base.TestCase): ENDPOINT_LIST_HEADERS = ['ID', 'Region', 'Service Name', 'Service Type', 'Enabled', 'Interface', 'URL'] - IDENTITY_PROVIDER_FIELDS = ['description', 'enabled', 'id', 'remote_ids'] + IDENTITY_PROVIDER_FIELDS = ['description', 'enabled', 'id', 'remote_ids', + 'domain_id'] IDENTITY_PROVIDER_LIST_HEADERS = ['ID', 'Enabled', 'Description'] SERVICE_PROVIDER_FIELDS = ['auth_url', 'description', 'enabled', diff --git a/openstackclient/tests/functional/identity/v3/test_idp.py b/openstackclient/tests/functional/identity/v3/test_idp.py index f9d8cb80..5db3610a 100644 --- a/openstackclient/tests/functional/identity/v3/test_idp.py +++ b/openstackclient/tests/functional/identity/v3/test_idp.py @@ -10,9 +10,10 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.identity.v3 import common from tempest.lib.common.utils import data_utils +from openstackclient.tests.functional.identity.v3 import common + class IdentityProviderTests(common.IdentityTests): # Introduce functional test case for command 'Identity Provider' diff --git a/openstackclient/tests/functional/image/v2/test_image.py b/openstackclient/tests/functional/image/v2/test_image.py index 3f432b02..6faff94a 100644 --- a/openstackclient/tests/functional/image/v2/test_image.py +++ b/openstackclient/tests/functional/image/v2/test_image.py @@ -74,3 +74,25 @@ class ImageTests(base.TestCase): self.openstack('image unset --property a --property c ' + self.NAME) raw_output = self.openstack('image show ' + self.NAME + opts) self.assertEqual(self.NAME + "\n\n", raw_output) + + def test_image_members(self): + opts = self.get_opts(['project_id']) + my_project_id = self.openstack('token issue' + opts).strip() + self.openstack( + 'image add project {} {}'.format(self.NAME, my_project_id)) + + self.openstack( + 'image set --accept ' + self.NAME) + shared_img_list = self.parse_listing( + self.openstack('image list --shared', self.get_opts(['name'])) + ) + self.assertIn(self.NAME, [img['Name'] for img in shared_img_list]) + + self.openstack( + 'image set --reject ' + self.NAME) + shared_img_list = self.parse_listing( + self.openstack('image list --shared', self.get_opts(['name'])) + ) + + self.openstack( + 'image remove project {} {}'.format(self.NAME, my_project_id)) 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), + ) diff --git a/openstackclient/tests/functional/post_test_hook_tips.sh b/openstackclient/tests/functional/post_test_hook_tips.sh new file mode 100755 index 00000000..28ab9580 --- /dev/null +++ b/openstackclient/tests/functional/post_test_hook_tips.sh @@ -0,0 +1,47 @@ +#!/bin/bash + +# This is a script that kicks off a series of functional tests against an +# OpenStack cloud. It will attempt to create an instance if one is not +# available. Do not run this script unless you know what you're doing. +# For more information refer to: +# http://docs.openstack.org/developer/python-openstackclient/ + +# This particular script differs from the normal post_test_hook because +# it installs the master (tip) version of osc-lib, os-client-config +# and openstacksdk, OSCs most important dependencies. + +function generate_testr_results { + if [ -f .testrepository/0 ]; then + sudo .tox/functional-tips/bin/testr last --subunit > $WORKSPACE/testrepository.subunit + sudo mv $WORKSPACE/testrepository.subunit $BASE/logs/testrepository.subunit + sudo .tox/functional-tips/bin/subunit2html $BASE/logs/testrepository.subunit $BASE/logs/testr_results.html + sudo gzip -9 $BASE/logs/testrepository.subunit + sudo gzip -9 $BASE/logs/testr_results.html + sudo chown jenkins:jenkins $BASE/logs/testrepository.subunit.gz $BASE/logs/testr_results.html.gz + sudo chmod a+r $BASE/logs/testrepository.subunit.gz $BASE/logs/testr_results.html.gz + fi +} + +export OPENSTACKCLIENT_DIR="$BASE/new/python-openstackclient" +sudo chown -R jenkins:stack $OPENSTACKCLIENT_DIR + +# Go to the openstackclient dir +cd $OPENSTACKCLIENT_DIR + +# Run tests +echo "Running openstackclient functional-tips test suite" +set +e + +# Source environment variables to kick things off +source ~stack/devstack/openrc admin admin +echo 'Running tests with:' +env | grep OS + +# Preserve env for OS_ credentials +sudo -E -H -u jenkins tox -e functional-tips +EXIT_CODE=$? +set -e + +# Collect and parse result +generate_testr_results +exit $EXIT_CODE diff --git a/openstackclient/tests/functional/volume/v2/test_qos.py b/openstackclient/tests/functional/volume/v2/test_qos.py index a54acbfd..1558c216 100644 --- a/openstackclient/tests/functional/volume/v2/test_qos.py +++ b/openstackclient/tests/functional/volume/v2/test_qos.py @@ -50,13 +50,13 @@ class QosTests(common.BaseVolumeTests): raw_output = self.openstack( 'volume qos set --property a=b --property c=d ' + self.ID) self.assertEqual("", raw_output) - opts = self.get_opts(['name', 'specs']) + opts = self.get_opts(['name', 'properties']) raw_output = self.openstack('volume qos show ' + self.ID + opts) self.assertEqual(self.NAME + "\na='b', c='d'\n", raw_output) raw_output = self.openstack( 'volume qos unset --property a ' + self.ID) self.assertEqual("", raw_output) - opts = self.get_opts(['name', 'specs']) + opts = self.get_opts(['name', 'properties']) raw_output = self.openstack('volume qos show ' + self.ID + opts) self.assertEqual(self.NAME + "\nc='d'\n", raw_output) diff --git a/openstackclient/tests/functional/volume/v2/test_snapshot.py b/openstackclient/tests/functional/volume/v2/test_snapshot.py index 4eb69e9d..c83e79f8 100644 --- a/openstackclient/tests/functional/volume/v2/test_snapshot.py +++ b/openstackclient/tests/functional/volume/v2/test_snapshot.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import json import time import uuid @@ -20,9 +21,6 @@ class VolumeSnapshotTests(common.BaseVolumeTests): """Functional tests for volume snapshot. """ VOLLY = uuid.uuid4().hex - NAME = uuid.uuid4().hex - OTHER_NAME = uuid.uuid4().hex - HEADERS = ['"Name"'] @classmethod def wait_for_status(cls, command, status, tries): @@ -37,52 +35,211 @@ class VolumeSnapshotTests(common.BaseVolumeTests): @classmethod def setUpClass(cls): super(VolumeSnapshotTests, cls).setUpClass() - cls.openstack('volume create --size 1 ' + cls.VOLLY) - cls.wait_for_status('volume show ' + cls.VOLLY, 'available\n', 3) - opts = cls.get_opts(['status']) - raw_output = cls.openstack('volume snapshot create --volume ' + - cls.VOLLY + ' ' + cls.NAME + opts) - cls.assertOutput('creating\n', raw_output) - cls.wait_for_status( - 'volume snapshot show ' + cls.NAME, 'available\n', 3) + # create a volume for all tests to create snapshot + cmd_output = json.loads(cls.openstack( + 'volume create -f json ' + + '--size 1 ' + + cls.VOLLY + )) + cls.wait_for_status('volume show ' + cls.VOLLY, 'available\n', 6) + cls.VOLUME_ID = cmd_output['id'] @classmethod def tearDownClass(cls): - # Rename test - raw_output = cls.openstack( - 'volume snapshot set --name ' + cls.OTHER_NAME + ' ' + cls.NAME) - cls.assertOutput('', raw_output) - # Delete test - raw_output_snapshot = cls.openstack( - 'volume snapshot delete ' + cls.OTHER_NAME) cls.wait_for_status('volume show ' + cls.VOLLY, 'available\n', 6) - raw_output_volume = cls.openstack('volume delete --force ' + cls.VOLLY) - cls.assertOutput('', raw_output_snapshot) - cls.assertOutput('', raw_output_volume) + raw_output = cls.openstack('volume delete --force ' + cls.VOLLY) + cls.assertOutput('', raw_output) - def test_snapshot_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('volume snapshot list' + opts) - self.assertIn(self.NAME, raw_output) + def test_volume_snapshot__delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name1 + + ' --volume ' + self.VOLLY + )) + self.assertEqual( + name1, + cmd_output["name"], + ) - def test_snapshot_properties(self): - raw_output = self.openstack( - 'volume snapshot set --property a=b --property c=d ' + self.NAME) - self.assertEqual("", raw_output) - opts = self.get_opts(["properties"]) - raw_output = self.openstack('volume snapshot show ' + self.NAME + opts) - self.assertEqual("a='b', c='d'\n", raw_output) + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name2 + + ' --volume ' + self.VOLLY + )) + self.assertEqual( + name2, + cmd_output["name"], + ) + + self.wait_for_status( + 'volume snapshot show ' + name1, 'available\n', 6) + self.wait_for_status( + 'volume snapshot show ' + name2, 'available\n', 6) + + del_output = self.openstack( + 'volume snapshot delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + + def test_volume_snapshot_list(self): + """Test create, list filter""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name1 + + ' --volume ' + self.VOLLY + )) + self.addCleanup(self.openstack, 'volume snapshot delete ' + name1) + self.assertEqual( + name1, + cmd_output["name"], + ) + self.assertEqual( + self.VOLUME_ID, + cmd_output["volume_id"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.wait_for_status( + 'volume snapshot show ' + name1, 'available\n', 6) + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name2 + + ' --volume ' + self.VOLLY + )) + self.addCleanup(self.openstack, 'volume snapshot delete ' + name2) + self.assertEqual( + name2, + cmd_output["name"], + ) + self.assertEqual( + self.VOLUME_ID, + cmd_output["volume_id"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.wait_for_status( + 'volume snapshot show ' + name2, 'available\n', 6) raw_output = self.openstack( - 'volume snapshot unset --property a ' + self.NAME) - self.assertEqual("", raw_output) - raw_output = self.openstack('volume snapshot show ' + self.NAME + opts) - self.assertEqual("c='d'\n", raw_output) + 'volume snapshot set ' + + '--state error ' + + name2 + ) + self.assertOutput('', raw_output) + + # Test list --long, --status + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--long ' + + '--status error' + )) + names = [x["Name"] for x in cmd_output] + self.assertNotIn(name1, names) + self.assertIn(name2, names) + + # Test list --volume + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--volume ' + self.VOLLY + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertIn(name2, names) + + # Test list --name + cmd_output = json.loads(self.openstack( + 'volume snapshot list -f json ' + + '--name ' + name1 + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertNotIn(name2, names) def test_snapshot_set(self): + """Test create, set, unset, show, delete volume snapshot""" + name = uuid.uuid4().hex + new_name = name + "_" + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + '--volume ' + self.VOLLY + + ' --description aaaa ' + + '--property Alpha=a ' + + name + )) + self.addCleanup(self.openstack, 'volume snapshot delete ' + new_name) + self.assertEqual( + name, + cmd_output["name"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='a'", + cmd_output["properties"], + ) + self.wait_for_status( + 'volume snapshot show ' + name, 'available\n', 6) + + # Test volume snapshot set raw_output = self.openstack( - 'volume snapshot set --description backup ' + self.NAME) - self.assertEqual("", raw_output) - opts = self.get_opts(["description", "name"]) - raw_output = self.openstack('volume snapshot show ' + self.NAME + opts) - self.assertEqual("backup\n" + self.NAME + "\n", raw_output) + 'volume snapshot set ' + + '--name ' + new_name + + ' --description bbbb ' + + '--property Alpha=c ' + + '--property Beta=b ' + + name, + ) + self.assertOutput('', raw_output) + + # Show snapshot set result + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertEqual( + new_name, + cmd_output["name"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='c', Beta='b'", + cmd_output["properties"], + ) + + # Test volume unset + raw_output = self.openstack( + 'volume snapshot unset ' + + '--property Alpha ' + + new_name, + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertEqual( + "Beta='b'", + cmd_output["properties"], + ) diff --git a/openstackclient/tests/functional/volume/v2/test_volume.py b/openstackclient/tests/functional/volume/v2/test_volume.py index ea891cba..203ca819 100644 --- a/openstackclient/tests/functional/volume/v2/test_volume.py +++ b/openstackclient/tests/functional/volume/v2/test_volume.py @@ -10,6 +10,7 @@ # License for the specific language governing permissions and limitations # under the License. +import json import time import uuid @@ -19,118 +20,220 @@ from openstackclient.tests.functional.volume.v2 import common class VolumeTests(common.BaseVolumeTests): """Functional tests for volume. """ - NAME = uuid.uuid4().hex - SNAPSHOT_NAME = uuid.uuid4().hex - VOLUME_FROM_SNAPSHOT_NAME = uuid.uuid4().hex - OTHER_NAME = uuid.uuid4().hex - HEADERS = ['"Display Name"'] - FIELDS = ['name'] - - @classmethod - def setUpClass(cls): - super(VolumeTests, cls).setUpClass() - opts = cls.get_opts(cls.FIELDS) - - # Create test volume - raw_output = cls.openstack('volume create --size 1 ' + cls.NAME + opts) - expected = cls.NAME + '\n' - cls.assertOutput(expected, raw_output) - - @classmethod - def tearDownClass(cls): - # Rename test volume - raw_output = cls.openstack( - 'volume set --name ' + cls.OTHER_NAME + ' ' + cls.NAME) - cls.assertOutput('', raw_output) - - # Set volume state - cls.openstack('volume set --state error ' + cls.OTHER_NAME) - opts = cls.get_opts(["status"]) - raw_output_status = cls.openstack( - 'volume show ' + cls.OTHER_NAME + opts) - - # Delete test volume - raw_output = cls.openstack('volume delete ' + cls.OTHER_NAME) - cls.assertOutput('', raw_output) - cls.assertOutput('error\n', raw_output_status) + def test_volume_delete(self): + """Test create, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + name1 + )) + self.assertEqual( + 1, + cmd_output["size"], + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 2 ' + + name2 + )) + self.assertEqual( + 2, + cmd_output["size"], + ) + + self.wait_for("volume", name1, "available") + self.wait_for("volume", name2, "available") + del_output = self.openstack('volume delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) def test_volume_list(self): - opts = self.get_opts(self.HEADERS) - raw_output = self.openstack('volume list' + opts) - self.assertIn(self.NAME, raw_output) - - def test_volume_show(self): - opts = self.get_opts(self.FIELDS) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n", raw_output) - - def test_volume_properties(self): + """Test create, list filter""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + name1 + )) + self.addCleanup(self.openstack, 'volume delete ' + name1) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.wait_for("volume", name1, "available") + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 2 ' + + name2 + )) + self.addCleanup(self.openstack, 'volume delete ' + name2) + self.assertEqual( + 2, + cmd_output["size"], + ) + self.wait_for("volume", name2, "available") raw_output = self.openstack( - 'volume set --property a=b --property c=d ' + self.NAME) - self.assertEqual("", raw_output) - opts = self.get_opts(["properties"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("a='b', c='d'\n", raw_output) + 'volume set ' + + '--state error ' + + name2 + ) + self.assertOutput('', raw_output) - raw_output = self.openstack('volume unset --property a ' + self.NAME) - self.assertEqual("", raw_output) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("c='d'\n", raw_output) + # Test list --long + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '--long' + )) + names = [x["Display Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertIn(name2, names) + + # Test list --status + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '--status error' + )) + names = [x["Display Name"] for x in cmd_output] + self.assertNotIn(name1, names) + self.assertIn(name2, names) + + # TODO(qiangjiahui): Add project option to filter tests when we can + # specify volume with project def test_volume_set(self): - discription = uuid.uuid4().hex - self.openstack('volume set --description ' + discription + ' ' + - self.NAME) - opts = self.get_opts(["description", "name"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(discription + "\n" + self.NAME + "\n", raw_output) - - def test_volume_set_size(self): - self.openstack('volume set --size 2 ' + self.NAME) - opts = self.get_opts(["name", "size"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual(self.NAME + "\n2\n", raw_output) - - def test_volume_set_bootable(self): - self.openstack('volume set --bootable ' + self.NAME) - opts = self.get_opts(["bootable"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("true\n", raw_output) - - self.openstack('volume set --non-bootable ' + self.NAME) - opts = self.get_opts(["bootable"]) - raw_output = self.openstack('volume show ' + self.NAME + opts) - self.assertEqual("false\n", raw_output) + """Tests create volume, set, unset, show, delete""" + name = uuid.uuid4().hex + new_name = name + "_" + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + '--description aaaa ' + + '--property Alpha=a ' + + name + )) + self.addCleanup(self.openstack, 'volume delete ' + new_name) + self.assertEqual( + name, + cmd_output["name"], + ) + self.assertEqual( + 1, + cmd_output["size"], + ) + self.assertEqual( + 'aaaa', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='a'", + cmd_output["properties"], + ) + self.assertEqual( + 'false', + cmd_output["bootable"], + ) + self.wait_for("volume", name, "available") + + # Test volume set + raw_output = self.openstack( + 'volume set ' + + '--name ' + new_name + + ' --size 2 ' + + '--description bbbb ' + + '--property Alpha=c ' + + '--property Beta=b ' + + '--bootable ' + + name, + ) + self.assertOutput('', raw_output) - def test_volume_snapshot(self): - opts = self.get_opts(self.FIELDS) - - # Create snapshot from test volume - raw_output = self.openstack('volume snapshot create ' + - self.SNAPSHOT_NAME + - ' --volume ' + self.NAME + opts) - expected = self.SNAPSHOT_NAME + '\n' - self.assertOutput(expected, raw_output) - self.wait_for("volume snapshot", self.SNAPSHOT_NAME, "available") - - # Create volume from snapshot - raw_output = self.openstack('volume create --size 2 --snapshot ' + - self.SNAPSHOT_NAME + ' ' + - self.VOLUME_FROM_SNAPSHOT_NAME + opts) - expected = self.VOLUME_FROM_SNAPSHOT_NAME + '\n' - self.assertOutput(expected, raw_output) - self.wait_for("volume", self.VOLUME_FROM_SNAPSHOT_NAME, "available") - - # Delete volume that create from snapshot - raw_output = self.openstack('volume delete ' + - self.VOLUME_FROM_SNAPSHOT_NAME) + cmd_output = json.loads(self.openstack( + 'volume show -f json ' + + new_name + )) + self.assertEqual( + new_name, + cmd_output["name"], + ) + self.assertEqual( + 2, + cmd_output["size"], + ) + self.assertEqual( + 'bbbb', + cmd_output["description"], + ) + self.assertEqual( + "Alpha='c', Beta='b'", + cmd_output["properties"], + ) + self.assertEqual( + 'true', + cmd_output["bootable"], + ) + + # Test volume unset + raw_output = self.openstack( + 'volume unset ' + + '--property Alpha ' + + new_name, + ) self.assertOutput('', raw_output) - # Delete test snapshot + cmd_output = json.loads(self.openstack( + 'volume show -f json ' + + new_name + )) + self.assertEqual( + "Beta='b'", + cmd_output["properties"], + ) + + def test_volume_snapshot(self): + """Tests volume create from snapshot""" + + volume_name = uuid.uuid4().hex + snapshot_name = uuid.uuid4().hex + # Make a snapshot + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + volume_name + )) + self.wait_for("volume", volume_name, "available") + self.assertEqual( + volume_name, + cmd_output["name"], + ) + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + snapshot_name + + ' --volume ' + volume_name + )) + self.wait_for("volume snapshot", snapshot_name, "available") + + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--snapshot ' + snapshot_name + + ' ' + name + )) + self.addCleanup(self.openstack, 'volume delete ' + name) + self.addCleanup(self.openstack, 'volume delete ' + volume_name) + self.assertEqual( + name, + cmd_output["name"], + ) + self.wait_for("volume", name, "available") + + # Delete snapshot raw_output = self.openstack( - 'volume snapshot delete ' + self.SNAPSHOT_NAME) + 'volume snapshot delete ' + snapshot_name) self.assertOutput('', raw_output) - self.wait_for("volume", self.NAME, "available") def wait_for(self, check_type, check_name, desired_status, wait=120, interval=5, failures=['ERROR']): diff --git a/openstackclient/tests/unit/api/test_image_v1.py b/openstackclient/tests/unit/api/test_image_v1.py index e02ef381..6ce3ddea 100644 --- a/openstackclient/tests/unit/api/test_image_v1.py +++ b/openstackclient/tests/unit/api/test_image_v1.py @@ -21,7 +21,7 @@ from openstackclient.tests.unit import utils FAKE_PROJECT = 'xyzpdq' -FAKE_URL = 'http://gopher.com' +FAKE_URL = 'http://gopher.dev10.com' class TestImageAPIv1(utils.TestCase): diff --git a/openstackclient/tests/unit/api/test_image_v2.py b/openstackclient/tests/unit/api/test_image_v2.py index 5dbb51e0..22490e46 100644 --- a/openstackclient/tests/unit/api/test_image_v2.py +++ b/openstackclient/tests/unit/api/test_image_v2.py @@ -21,7 +21,7 @@ from openstackclient.tests.unit import utils FAKE_PROJECT = 'xyzpdq' -FAKE_URL = 'http://gopher.com' +FAKE_URL = 'http://gopher.dev20.com' class TestImageAPIv2(utils.TestCase): diff --git a/openstackclient/tests/unit/common/test_quota.py b/openstackclient/tests/unit/common/test_quota.py index 7dd23373..244d74d2 100644 --- a/openstackclient/tests/unit/common/test_quota.py +++ b/openstackclient/tests/unit/common/test_quota.py @@ -13,6 +13,8 @@ import copy import mock +from openstack.network.v2 import quota as _quota + from openstackclient.common import quota from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes from openstackclient.tests.unit import fakes @@ -282,27 +284,32 @@ class TestQuotaSet(TestQuota): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - result = self.cmd.take_action(parsed_args) - kwargs = { - 'subnet': network_fakes.QUOTA['subnet'], - 'network': network_fakes.QUOTA['network'], - 'floatingip': network_fakes.QUOTA['floatingip'], - 'subnetpool': network_fakes.QUOTA['subnetpool'], - 'security_group_rule': - network_fakes.QUOTA['security_group_rule'], - 'security_group': network_fakes.QUOTA['security_group'], - 'router': network_fakes.QUOTA['router'], - 'rbac_policy': network_fakes.QUOTA['rbac_policy'], - 'port': network_fakes.QUOTA['port'], - 'vip': network_fakes.QUOTA['vip'], - 'healthmonitor': network_fakes.QUOTA['healthmonitor'], - 'l7policy': network_fakes.QUOTA['l7policy'], - } - self.network_mock.update_quota.assert_called_once_with( - identity_fakes.project_id, - **kwargs - ) - self.assertIsNone(result) + # TODO(huanxuan): Remove this if condition once the fixed + # SDK Quota class is the minimum required version. + # This is expected to be SDK release 0.9.13 + if not hasattr(_quota.Quota, 'allow_get'): + # Just run this when sdk <= 0.9.10 + result = self.cmd.take_action(parsed_args) + kwargs = { + 'subnet': network_fakes.QUOTA['subnet'], + 'network': network_fakes.QUOTA['network'], + 'floatingip': network_fakes.QUOTA['floatingip'], + 'subnetpool': network_fakes.QUOTA['subnetpool'], + 'security_group_rule': + network_fakes.QUOTA['security_group_rule'], + 'security_group': network_fakes.QUOTA['security_group'], + 'router': network_fakes.QUOTA['router'], + 'rbac_policy': network_fakes.QUOTA['rbac_policy'], + 'port': network_fakes.QUOTA['port'], + 'vip': network_fakes.QUOTA['vip'], + 'healthmonitor': network_fakes.QUOTA['healthmonitor'], + 'l7policy': network_fakes.QUOTA['l7policy'], + } + self.network_mock.update_quota.assert_called_once_with( + identity_fakes.project_id, + **kwargs + ) + self.assertIsNone(result) def test_quota_set_with_class(self): arglist = [ @@ -476,15 +483,20 @@ class TestQuotaShow(TestQuota): parsed_args = self.check_parser(self.cmd, arglist, verifylist) - self.cmd.take_action(parsed_args) - - self.quotas_mock.defaults.assert_called_once_with( - identity_fakes.project_id) - self.volume_quotas_mock.defaults.assert_called_once_with( - identity_fakes.project_id) - self.network.get_quota_default.assert_called_once_with( - identity_fakes.project_id) - self.assertNotCalled(self.network.get_quota) + # TODO(huanxuan): Remove this if condition once the fixed + # SDK QuotaDefault class is the minimum required version. + # This is expected to be SDK release 0.9.13 + if not hasattr(_quota.QuotaDefault, 'project'): + # Just run this when sdk <= 0.9.10 + self.cmd.take_action(parsed_args) + + self.quotas_mock.defaults.assert_called_once_with( + identity_fakes.project_id) + self.volume_quotas_mock.defaults.assert_called_once_with( + identity_fakes.project_id) + self.network.get_quota_default.assert_called_once_with( + identity_fakes.project_id) + self.assertNotCalled(self.network.get_quota) def test_quota_show_with_class(self): arglist = [ diff --git a/openstackclient/tests/unit/compute/v2/fakes.py b/openstackclient/tests/unit/compute/v2/fakes.py index 985ce5e2..4fe735b6 100644 --- a/openstackclient/tests/unit/compute/v2/fakes.py +++ b/openstackclient/tests/unit/compute/v2/fakes.py @@ -168,6 +168,9 @@ class FakeComputev2Client(object): self.quota_classes = mock.Mock() self.quota_classes.resource_class = fakes.FakeResource(None, {}) + self.usage = mock.Mock() + self.usage.resource_class = fakes.FakeResource(None, {}) + self.volumes = mock.Mock() self.volumes.resource_class = fakes.FakeResource(None, {}) @@ -1248,3 +1251,65 @@ class FakeServerGroup(object): info=copy.deepcopy(server_group_info), loaded=True) return server_group + + +class FakeUsage(object): + """Fake one or more usage.""" + + @staticmethod + def create_one_usage(attrs=None): + """Create a fake usage. + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object, with tenant_id and other attributes + """ + if attrs is None: + attrs = {} + + # Set default attributes. + usage_info = { + 'tenant_id': 'usage-tenant-id-' + uuid.uuid4().hex, + 'total_memory_mb_usage': 512.0, + 'total_vcpus_usage': 1.0, + 'total_local_gb_usage': 1.0, + 'server_usages': [ + { + 'ended_at': None, + 'flavor': 'usage-flavor-' + uuid.uuid4().hex, + 'hours': 1.0, + 'local_gb': 1, + 'memory_mb': 512, + 'name': 'usage-name-' + uuid.uuid4().hex, + 'state': 'active', + 'uptime': 3600, + 'vcpus': 1 + } + ] + } + + # Overwrite default attributes. + usage_info.update(attrs) + + usage = fakes.FakeResource(info=copy.deepcopy(usage_info), + loaded=True) + + return usage + + @staticmethod + def create_usages(attrs=None, count=2): + """Create multiple fake services. + + :param Dictionary attrs: + A dictionary with all attributes + :param int count: + The number of services to fake + :return: + A list of FakeResource objects faking the services + """ + usages = [] + for i in range(0, count): + usages.append(FakeUsage.create_one_usage(attrs)) + + return usages diff --git a/openstackclient/tests/unit/compute/v2/test_flavor.py b/openstackclient/tests/unit/compute/v2/test_flavor.py index 93ad9d14..632fcda1 100644 --- a/openstackclient/tests/unit/compute/v2/test_flavor.py +++ b/openstackclient/tests/unit/compute/v2/test_flavor.py @@ -160,7 +160,7 @@ class TestFlavorCreate(TestFlavor): self.flavor.is_public = False arglist = [ - '--id', self.flavor.id, + '--id', 'auto', '--ram', str(self.flavor.ram), '--disk', str(self.flavor.disk), '--ephemeral', str(self.flavor.ephemeral), @@ -174,7 +174,6 @@ class TestFlavorCreate(TestFlavor): self.flavor.name, ] verifylist = [ - ('id', self.flavor.id), ('ram', self.flavor.ram), ('disk', self.flavor.disk), ('ephemeral', self.flavor.ephemeral), @@ -193,7 +192,7 @@ class TestFlavorCreate(TestFlavor): self.flavor.ram, self.flavor.vcpus, self.flavor.disk, - self.flavor.id, + 'auto', self.flavor.ephemeral, self.flavor.swap, self.flavor.rxtx_factor, diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py index 57221189..54f36209 100644 --- a/openstackclient/tests/unit/compute/v2/test_server.py +++ b/openstackclient/tests/unit/compute/v2/test_server.py @@ -36,10 +36,6 @@ class TestServer(compute_fakes.TestComputev2): self.servers_mock = self.app.client_manager.compute.servers self.servers_mock.reset_mock() - # Get a shortcut to the compute client ImageManager Mock - self.cimages_mock = self.app.client_manager.compute.images - self.cimages_mock.reset_mock() - # Get a shortcut to the compute client FlavorManager Mock self.flavors_mock = self.app.client_manager.compute.flavors self.flavors_mock.reset_mock() @@ -259,7 +255,7 @@ class TestServerCreate(TestServer): self.servers_mock.create.return_value = self.new_server self.image = image_fakes.FakeImage.create_one_image() - self.cimages_mock.get.return_value = self.image + self.images_mock.get.return_value = self.image self.flavor = compute_fakes.FakeFlavor.create_one_flavor() self.flavors_mock.get.return_value = self.flavor @@ -859,7 +855,7 @@ class TestServerList(TestServer): self.servers_mock.list.return_value = self.servers self.image = image_fakes.FakeImage.create_one_image() - self.cimages_mock.get.return_value = self.image + self.images_mock.get.return_value = self.image self.flavor = compute_fakes.FakeFlavor.create_one_flavor() self.flavors_mock.get.return_value = self.flavor @@ -943,7 +939,7 @@ class TestServerList(TestServer): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) - self.cimages_mock.get.assert_any_call(self.image.id) + self.images_mock.get.assert_any_call(self.image.id) self.search_opts['image'] = self.image.id self.servers_mock.list.assert_called_with(**self.kwargs) @@ -1019,7 +1015,7 @@ class TestServerRebuild(TestServer): # Return value for utils.find_resource for image self.image = image_fakes.FakeImage.create_one_image() - self.cimages_mock.get.return_value = self.image + self.images_mock.get.return_value = self.image # Fake the rebuilt new server. new_server = compute_fakes.FakeServer.create_one_server() @@ -1059,7 +1055,7 @@ class TestServerRebuild(TestServer): self.cmd.take_action(parsed_args) self.servers_mock.get.assert_called_with(self.server.id) - self.cimages_mock.get.assert_called_with(self.image.id) + self.images_mock.get.assert_called_with(self.image.id) self.server.rebuild.assert_called_with(self.image, None) def test_rebuild_with_current_image_and_password(self): @@ -1078,7 +1074,7 @@ class TestServerRebuild(TestServer): self.cmd.take_action(parsed_args) self.servers_mock.get.assert_called_with(self.server.id) - self.cimages_mock.get.assert_called_with(self.image.id) + self.images_mock.get.assert_called_with(self.image.id) self.server.rebuild.assert_called_with(self.image, password) @mock.patch.object(common_utils, 'wait_for_status', return_value=True) @@ -1106,7 +1102,7 @@ class TestServerRebuild(TestServer): ) self.servers_mock.get.assert_called_with(self.server.id) - self.cimages_mock.get.assert_called_with(self.image.id) + self.images_mock.get.assert_called_with(self.image.id) self.server.rebuild.assert_called_with(self.image, None) @mock.patch.object(common_utils, 'wait_for_status', return_value=False) @@ -1130,7 +1126,7 @@ class TestServerRebuild(TestServer): ) self.servers_mock.get.assert_called_with(self.server.id) - self.cimages_mock.get.assert_called_with(self.image.id) + self.images_mock.get.assert_called_with(self.image.id) self.server.rebuild.assert_called_with(self.image, None) @@ -1628,7 +1624,7 @@ class TestServerShow(TestServer): # This is the return value for utils.find_resource() self.servers_mock.get.return_value = self.server - self.cimages_mock.get.return_value = self.image + self.images_mock.get.return_value = self.image self.flavors_mock.get.return_value = self.flavor # Get the command object to test @@ -1986,6 +1982,7 @@ class TestServerGeneral(TestServer): # Call _prep_server_detail(). server_detail = server._prep_server_detail( self.app.client_manager.compute, + self.app.client_manager.image, _server ) # 'networks' is used to create _server. Remove it. diff --git a/openstackclient/tests/unit/compute/v2/test_usage.py b/openstackclient/tests/unit/compute/v2/test_usage.py new file mode 100644 index 00000000..a383e903 --- /dev/null +++ b/openstackclient/tests/unit/compute/v2/test_usage.py @@ -0,0 +1,179 @@ +# 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 datetime +import mock + +from openstackclient.compute.v2 import usage +from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes +from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes + + +class TestUsage(compute_fakes.TestComputev2): + + def setUp(self): + super(TestUsage, self).setUp() + + self.usage_mock = self.app.client_manager.compute.usage + self.usage_mock.reset_mock() + + self.projects_mock = self.app.client_manager.identity.projects + self.projects_mock.reset_mock() + + +class TestUsageList(TestUsage): + + project = identity_fakes.FakeProject.create_one_project() + # Return value of self.usage_mock.list(). + usages = compute_fakes.FakeUsage.create_usages( + attrs={'tenant_id': project.name}, count=1) + + columns = ( + "Project", + "Servers", + "RAM MB-Hours", + "CPU Hours", + "Disk GB-Hours" + ) + + data = [( + usages[0].tenant_id, + len(usages[0].server_usages), + float("%.2f" % usages[0].total_memory_mb_usage), + float("%.2f" % usages[0].total_vcpus_usage), + float("%.2f" % usages[0].total_local_gb_usage), + )] + + def setUp(self): + super(TestUsageList, self).setUp() + + self.usage_mock.list.return_value = self.usages + + self.projects_mock.list.return_value = [self.project] + # Get the command object to test + self.cmd = usage.ListUsage(self.app, None) + + def test_usage_list_no_options(self): + + arglist = [] + verifylist = [ + ('start', None), + ('end', None), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.projects_mock.list.assert_called_with() + + self.assertEqual(self.columns, columns) + self.assertEqual(tuple(self.data), tuple(data)) + + def test_usage_list_with_options(self): + arglist = [ + '--start', '2016-11-11', + '--end', '2016-12-20', + ] + verifylist = [ + ('start', '2016-11-11'), + ('end', '2016-12-20'), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.projects_mock.list.assert_called_with() + self.usage_mock.list.assert_called_with( + datetime.datetime(2016, 11, 11, 0, 0), + datetime.datetime(2016, 12, 20, 0, 0), + detailed=True) + + self.assertEqual(self.columns, columns) + self.assertEqual(tuple(self.data), tuple(data)) + + +class TestUsageShow(TestUsage): + + project = identity_fakes.FakeProject.create_one_project() + # Return value of self.usage_mock.list(). + usage = compute_fakes.FakeUsage.create_one_usage( + attrs={'tenant_id': project.name}) + + columns = ( + 'CPU Hours', + 'Disk GB-Hours', + 'RAM MB-Hours', + 'Servers', + ) + + data = ( + float("%.2f" % usage.total_vcpus_usage), + float("%.2f" % usage.total_local_gb_usage), + float("%.2f" % usage.total_memory_mb_usage), + len(usage.server_usages), + ) + + def setUp(self): + super(TestUsageShow, self).setUp() + + self.usage_mock.get.return_value = self.usage + + self.projects_mock.get.return_value = self.project + # Get the command object to test + self.cmd = usage.ShowUsage(self.app, None) + + def test_usage_show_no_options(self): + + self.app.client_manager.auth_ref = mock.Mock() + self.app.client_manager.auth_ref.project_id = self.project.id + + arglist = [] + verifylist = [ + ('project', None), + ('start', None), + ('end', None), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_usage_show_with_options(self): + + arglist = [ + '--project', self.project.id, + '--start', '2016-11-11', + '--end', '2016-12-20', + ] + verifylist = [ + ('project', self.project.id), + ('start', '2016-11-11'), + ('end', '2016-12-20'), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.usage_mock.get.assert_called_with( + self.project.id, + datetime.datetime(2016, 11, 11, 0, 0), + datetime.datetime(2016, 12, 20, 0, 0)) + + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/fakes.py b/openstackclient/tests/unit/fakes.py index f7cb5676..626b466d 100644 --- a/openstackclient/tests/unit/fakes.py +++ b/openstackclient/tests/unit/fakes.py @@ -212,10 +212,19 @@ class FakeResource(object): def keys(self): return self._info.keys() + def to_dict(self): + return self._info + @property def info(self): return self._info + def __getitem__(self, item): + return self._info.get(item) + + def get(self, item, default=None): + return self._info.get(item, default) + class FakeResponse(requests.Response): diff --git a/openstackclient/tests/unit/identity/v2_0/test_project.py b/openstackclient/tests/unit/identity/v2_0/test_project.py index c1f00762..4e1077db 100644 --- a/openstackclient/tests/unit/identity/v2_0/test_project.py +++ b/openstackclient/tests/unit/identity/v2_0/test_project.py @@ -13,8 +13,11 @@ # under the License. # +import mock + from keystoneauth1 import exceptions as ks_exc from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v2_0 import project from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes @@ -302,6 +305,32 @@ class TestProjectDelete(TestProject): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_projects_with_exception(self, find_mock): + find_mock.side_effect = [self.fake_project, + exceptions.CommandError] + arglist = [ + self.fake_project.id, + 'unexist_project', + ] + verifylist = [ + ('projects', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 projects failed to delete.', + str(e)) + + find_mock.assert_any_call(self.projects_mock, self.fake_project.id) + find_mock.assert_any_call(self.projects_mock, 'unexist_project') + + self.assertEqual(2, find_mock.call_count) + self.projects_mock.delete.assert_called_once_with(self.fake_project.id) + class TestProjectList(TestProject): diff --git a/openstackclient/tests/unit/identity/v2_0/test_role.py b/openstackclient/tests/unit/identity/v2_0/test_role.py index 68ebf141..684ce803 100644 --- a/openstackclient/tests/unit/identity/v2_0/test_role.py +++ b/openstackclient/tests/unit/identity/v2_0/test_role.py @@ -17,6 +17,7 @@ import mock from keystoneauth1 import exceptions as ks_exc from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v2_0 import role from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes @@ -240,6 +241,32 @@ class TestRoleDelete(TestRole): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_roles_with_exception(self, find_mock): + find_mock.side_effect = [self.fake_role, + exceptions.CommandError] + arglist = [ + self.fake_role.id, + 'unexist_role', + ] + verifylist = [ + ('roles', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 roles failed to delete.', + str(e)) + + find_mock.assert_any_call(self.roles_mock, self.fake_role.id) + find_mock.assert_any_call(self.roles_mock, 'unexist_role') + + self.assertEqual(2, find_mock.call_count) + self.roles_mock.delete.assert_called_once_with(self.fake_role.id) + class TestRoleList(TestRole): diff --git a/openstackclient/tests/unit/identity/v2_0/test_user.py b/openstackclient/tests/unit/identity/v2_0/test_user.py index 765f8559..a8b9497e 100644 --- a/openstackclient/tests/unit/identity/v2_0/test_user.py +++ b/openstackclient/tests/unit/identity/v2_0/test_user.py @@ -17,6 +17,7 @@ import mock from keystoneauth1 import exceptions as ks_exc from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v2_0 import user from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes @@ -411,6 +412,32 @@ class TestUserDelete(TestUser): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_users_with_exception(self, find_mock): + find_mock.side_effect = [self.fake_user, + exceptions.CommandError] + arglist = [ + self.fake_user.id, + 'unexist_user', + ] + verifylist = [ + ('users', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 users failed to delete.', + str(e)) + + find_mock.assert_any_call(self.users_mock, self.fake_user.id) + find_mock.assert_any_call(self.users_mock, 'unexist_user') + + self.assertEqual(2, find_mock.call_count) + self.users_mock.delete.assert_called_once_with(self.fake_user.id) + class TestUserList(TestUser): diff --git a/openstackclient/tests/unit/identity/v3/test_group.py b/openstackclient/tests/unit/identity/v3/test_group.py index eb50adb5..8558de95 100644 --- a/openstackclient/tests/unit/identity/v3/test_group.py +++ b/openstackclient/tests/unit/identity/v3/test_group.py @@ -16,6 +16,7 @@ from mock import call from keystoneauth1 import exceptions as ks_exc from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v3 import group from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes @@ -257,6 +258,32 @@ class TestGroupDelete(TestGroup): self.groups_mock.delete.assert_called_once_with(self.groups[0].id) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_groups_with_exception(self, find_mock): + find_mock.side_effect = [self.groups[0], + exceptions.CommandError] + arglist = [ + self.groups[0].id, + 'unexist_group', + ] + verifylist = [ + ('groups', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 groups failed to delete.', + str(e)) + + find_mock.assert_any_call(self.groups_mock, self.groups[0].id) + find_mock.assert_any_call(self.groups_mock, 'unexist_group') + + self.assertEqual(2, find_mock.call_count) + self.groups_mock.delete.assert_called_once_with(self.groups[0].id) + class TestGroupList(TestGroup): diff --git a/openstackclient/tests/unit/identity/v3/test_mappings.py b/openstackclient/tests/unit/identity/v3/test_mappings.py index 5086724c..93fe1196 100644 --- a/openstackclient/tests/unit/identity/v3/test_mappings.py +++ b/openstackclient/tests/unit/identity/v3/test_mappings.py @@ -181,16 +181,12 @@ class TestMappingSet(TestMapping): mocker.return_value = identity_fakes.MAPPING_RULES_2 with mock.patch("openstackclient.identity.v3.mapping." "SetMapping._read_rules", mocker): - columns, data = self.cmd.take_action(parsed_args) + result = self.cmd.take_action(parsed_args) self.mapping_mock.update.assert_called_with( mapping=identity_fakes.mapping_id, rules=identity_fakes.MAPPING_RULES_2) - collist = ('id', 'rules') - self.assertEqual(collist, columns) - datalist = (identity_fakes.mapping_id, - identity_fakes.MAPPING_RULES_2) - self.assertEqual(datalist, data) + self.assertIsNone(result) def test_set_rules_wrong_file_path(self): arglist = [ diff --git a/openstackclient/tests/unit/identity/v3/test_project.py b/openstackclient/tests/unit/identity/v3/test_project.py index 702d9209..2b898090 100644 --- a/openstackclient/tests/unit/identity/v3/test_project.py +++ b/openstackclient/tests/unit/identity/v3/test_project.py @@ -16,6 +16,7 @@ import mock from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v3 import project from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes @@ -445,6 +446,32 @@ class TestProjectDelete(TestProject): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_projects_with_exception(self, find_mock): + find_mock.side_effect = [self.project, + exceptions.CommandError] + arglist = [ + self.project.id, + 'unexist_project', + ] + verifylist = [ + ('projects', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 projects failed to delete.', + str(e)) + + find_mock.assert_any_call(self.projects_mock, self.project.id) + find_mock.assert_any_call(self.projects_mock, 'unexist_project') + + self.assertEqual(2, find_mock.call_count) + self.projects_mock.delete.assert_called_once_with(self.project.id) + class TestProjectList(TestProject): diff --git a/openstackclient/tests/unit/identity/v3/test_role.py b/openstackclient/tests/unit/identity/v3/test_role.py index 448e18d3..c0b68bdf 100644 --- a/openstackclient/tests/unit/identity/v3/test_role.py +++ b/openstackclient/tests/unit/identity/v3/test_role.py @@ -14,6 +14,10 @@ # import copy +import mock + +from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v3 import role from openstackclient.tests.unit import fakes @@ -428,6 +432,36 @@ class TestRoleDelete(TestRole): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_roles_with_exception(self, find_mock): + find_mock.side_effect = [self.roles_mock.get.return_value, + exceptions.CommandError] + arglist = [ + identity_fakes.role_name, + 'unexist_role', + ] + verifylist = [ + ('roles', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 roles failed to delete.', + str(e)) + + find_mock.assert_any_call(self.roles_mock, + identity_fakes.role_name, + domain_id=None) + find_mock.assert_any_call(self.roles_mock, + 'unexist_role', + domain_id=None) + + self.assertEqual(2, find_mock.call_count) + self.roles_mock.delete.assert_called_once_with(identity_fakes.role_id) + class TestRoleList(TestRole): diff --git a/openstackclient/tests/unit/identity/v3/test_trust.py b/openstackclient/tests/unit/identity/v3/test_trust.py index 4eeb8bfe..93e8f63d 100644 --- a/openstackclient/tests/unit/identity/v3/test_trust.py +++ b/openstackclient/tests/unit/identity/v3/test_trust.py @@ -12,6 +12,10 @@ # import copy +import mock + +from osc_lib import exceptions +from osc_lib import utils from openstackclient.identity.v3 import trust from openstackclient.tests.unit import fakes @@ -148,6 +152,33 @@ class TestTrustDelete(TestTrust): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_trusts_with_exception(self, find_mock): + find_mock.side_effect = [self.trusts_mock.get.return_value, + exceptions.CommandError] + arglist = [ + identity_fakes.trust_id, + 'unexist_trust', + ] + verifylist = [ + ('trust', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 trusts failed to delete.', + str(e)) + + find_mock.assert_any_call(self.trusts_mock, identity_fakes.trust_id) + find_mock.assert_any_call(self.trusts_mock, 'unexist_trust') + + self.assertEqual(2, find_mock.call_count) + self.trusts_mock.delete.assert_called_once_with( + identity_fakes.trust_id) + class TestTrustList(TestTrust): diff --git a/openstackclient/tests/unit/identity/v3/test_user.py b/openstackclient/tests/unit/identity/v3/test_user.py index 6150a5f3..3c1f49a6 100644 --- a/openstackclient/tests/unit/identity/v3/test_user.py +++ b/openstackclient/tests/unit/identity/v3/test_user.py @@ -16,6 +16,9 @@ import contextlib import mock +from osc_lib import exceptions +from osc_lib import utils + from openstackclient.identity.v3 import user from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes @@ -465,6 +468,32 @@ class TestUserDelete(TestUser): ) self.assertIsNone(result) + @mock.patch.object(utils, 'find_resource') + def test_delete_multi_users_with_exception(self, find_mock): + find_mock.side_effect = [self.user, + exceptions.CommandError] + arglist = [ + self.user.id, + 'unexist_user', + ] + verifylist = [ + ('users', arglist), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 users failed to delete.', + str(e)) + + find_mock.assert_any_call(self.users_mock, self.user.id) + find_mock.assert_any_call(self.users_mock, 'unexist_user') + + self.assertEqual(2, find_mock.call_count) + self.users_mock.delete.assert_called_once_with(self.user.id) + class TestUserList(TestUser): diff --git a/openstackclient/tests/unit/image/v2/test_image.py b/openstackclient/tests/unit/image/v2/test_image.py index a054e513..164185df 100644 --- a/openstackclient/tests/unit/image/v2/test_image.py +++ b/openstackclient/tests/unit/image/v2/test_image.py @@ -829,6 +829,11 @@ class TestImageSet(TestImage): self.images_mock.get.return_value = self.model(**image_fakes.IMAGE) self.images_mock.update.return_value = self.model(**image_fakes.IMAGE) + + self.app.client_manager.auth_ref = mock.Mock( + project_id=self.project.id, + ) + # Get the command object to test self.cmd = image.SetImage(self.app, None) @@ -845,6 +850,101 @@ class TestImageSet(TestImage): self.assertIsNone(result) + self.image_members_mock.update.assert_not_called() + + def test_image_set_membership_option_accept(self): + membership = image_fakes.FakeImage.create_one_image_member( + attrs={'image_id': image_fakes.image_id, + 'member_id': self.project.id} + ) + self.image_members_mock.update.return_value = membership + + arglist = [ + '--accept', + image_fakes.image_id, + ] + verifylist = [ + ('accept', True), + ('reject', False), + ('pending', False), + ('image', image_fakes.image_id) + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + + self.image_members_mock.update.assert_called_once_with( + image_fakes.image_id, + self.app.client_manager.auth_ref.project_id, + 'accepted', + ) + + # Assert that the 'update image" route is also called, in addition to + # the 'update membership' route. + self.images_mock.update.assert_called_with(image_fakes.image_id) + + def test_image_set_membership_option_reject(self): + membership = image_fakes.FakeImage.create_one_image_member( + attrs={'image_id': image_fakes.image_id, + 'member_id': self.project.id} + ) + self.image_members_mock.update.return_value = membership + + arglist = [ + '--reject', + image_fakes.image_id, + ] + verifylist = [ + ('accept', False), + ('reject', True), + ('pending', False), + ('image', image_fakes.image_id) + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + + self.image_members_mock.update.assert_called_once_with( + image_fakes.image_id, + self.app.client_manager.auth_ref.project_id, + 'rejected', + ) + + # Assert that the 'update image" route is also called, in addition to + # the 'update membership' route. + self.images_mock.update.assert_called_with(image_fakes.image_id) + + def test_image_set_membership_option_pending(self): + membership = image_fakes.FakeImage.create_one_image_member( + attrs={'image_id': image_fakes.image_id, + 'member_id': self.project.id} + ) + self.image_members_mock.update.return_value = membership + + arglist = [ + '--pending', + image_fakes.image_id, + ] + verifylist = [ + ('accept', False), + ('reject', False), + ('pending', True), + ('image', image_fakes.image_id) + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + + self.image_members_mock.update.assert_called_once_with( + image_fakes.image_id, + self.app.client_manager.auth_ref.project_id, + 'pending', + ) + + # Assert that the 'update image" route is also called, in addition to + # the 'update membership' route. + self.images_mock.update.assert_called_with(image_fakes.image_id) + def test_image_set_options(self): arglist = [ '--name', 'new-name', diff --git a/openstackclient/tests/unit/network/v2/fakes.py b/openstackclient/tests/unit/network/v2/fakes.py index 84f145fb..524285ab 100644 --- a/openstackclient/tests/unit/network/v2/fakes.py +++ b/openstackclient/tests/unit/network/v2/fakes.py @@ -14,6 +14,8 @@ import argparse import copy import mock +from random import choice +from random import randint import uuid from openstackclient.tests.unit import fakes @@ -37,10 +39,20 @@ QUOTA = { "l7policy": 5, } +RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit' +RULE_TYPE_DSCP_MARKING = 'dscp-marking' +RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth' +VALID_QOS_RULES = [RULE_TYPE_BANDWIDTH_LIMIT, + RULE_TYPE_DSCP_MARKING, + RULE_TYPE_MINIMUM_BANDWIDTH] +VALID_DSCP_MARKS = [0, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, + 34, 36, 38, 40, 46, 48, 56] + class FakeNetworkV2Client(object): def __init__(self, **kwargs): + self.session = mock.Mock() self.extensions = mock.Mock() self.extensions.resource_class = fakes.FakeResource(None, {}) @@ -553,6 +565,8 @@ class FakeNetworkAgent(object): agent_attrs.update(attrs) agent = fakes.FakeResource(info=copy.deepcopy(agent_attrs), loaded=True) + agent.is_admin_state_up = agent_attrs['admin_state_up'] + agent.is_alive = agent_attrs['alive'] return agent @staticmethod @@ -662,83 +676,90 @@ class FakeNetworkRBAC(object): return mock.Mock(side_effect=rbac_policies) -class FakeNetworkQosBandwidthLimitRule(object): - """Fake one or more QoS bandwidth limit rules.""" +class FakeNetworkQosPolicy(object): + """Fake one or more QoS policies.""" @staticmethod - def create_one_qos_bandwidth_limit_rule(attrs=None): - """Create a fake QoS bandwidth limit rule. + def create_one_qos_policy(attrs=None): + """Create a fake QoS policy. :param Dictionary attrs: A dictionary with all attributes :return: - A FakeResource object with id, qos_policy_id, max_kbps and - max_burst_kbps attributes. + A FakeResource object with name, id, etc. """ attrs = attrs or {} + qos_id = attrs.get('id') or 'qos-policy-id-' + uuid.uuid4().hex + rule_attrs = {'qos_policy_id': qos_id} + rules = [FakeNetworkQosRule.create_one_qos_rule(rule_attrs)] # Set default attributes. - qos_bandwidth_limit_rule_attrs = { - 'id': 'qos-bandwidth-limit-rule-id-' + uuid.uuid4().hex, - 'qos_policy_id': 'qos-policy-id-' + uuid.uuid4().hex, - 'max_kbps': 1500, - 'max_burst_kbps': 1200, + qos_policy_attrs = { + 'name': 'qos-policy-name-' + uuid.uuid4().hex, + 'id': qos_id, + 'tenant_id': 'project-id-' + uuid.uuid4().hex, + 'shared': False, + 'description': 'qos-policy-description-' + uuid.uuid4().hex, + 'rules': rules, } # Overwrite default attributes. - qos_bandwidth_limit_rule_attrs.update(attrs) + qos_policy_attrs.update(attrs) - qos_bandwidth_limit_rule = fakes.FakeResource( - info=copy.deepcopy(qos_bandwidth_limit_rule_attrs), + qos_policy = fakes.FakeResource( + info=copy.deepcopy(qos_policy_attrs), loaded=True) - return qos_bandwidth_limit_rule + # Set attributes with special mapping in OpenStack SDK. + qos_policy.is_shared = qos_policy_attrs['shared'] + qos_policy.project_id = qos_policy_attrs['tenant_id'] + + return qos_policy @staticmethod - def create_qos_bandwidth_limit_rules(attrs=None, count=2): - """Create multiple fake QoS bandwidth limit rules. + def create_qos_policies(attrs=None, count=2): + """Create multiple fake QoS policies. :param Dictionary attrs: A dictionary with all attributes :param int count: - The number of QoS bandwidth limit rules to fake + The number of QoS policies to fake :return: - A list of FakeResource objects faking the QoS bandwidth limit rules + A list of FakeResource objects faking the QoS policies """ qos_policies = [] for i in range(0, count): - qos_policies.append(FakeNetworkQosBandwidthLimitRule. - create_one_qos_bandwidth_limit_rule(attrs)) + qos_policies.append( + FakeNetworkQosPolicy.create_one_qos_policy(attrs)) return qos_policies @staticmethod - def get_qos_bandwidth_limit_rules(qos_rules=None, count=2): - """Get a list of faked QoS bandwidth limit rules. + def get_qos_policies(qos_policies=None, count=2): + """Get an iterable MagicMock object with a list of faked QoS policies. - If QoS bandwidth limit rules list is provided, then initialize the - Mock object with the list. Otherwise create one. + If qos policies list is provided, then initialize the Mock object + with the list. Otherwise create one. :param List address scopes: - A list of FakeResource objects faking QoS bandwidth limit rules + A list of FakeResource objects faking qos policies :param int count: - The number of QoS bandwidth limit rules to fake + The number of QoS policies to fake :return: An iterable Mock object with side_effect set to a list of faked - qos bandwidth limit rules + QoS policies """ - if qos_rules is None: - qos_rules = (FakeNetworkQosBandwidthLimitRule. - create_qos_bandwidth_limit_rules(count)) - return mock.Mock(side_effect=qos_rules) + if qos_policies is None: + qos_policies = FakeNetworkQosPolicy.create_qos_policies(count) + return mock.Mock(side_effect=qos_policies) -class FakeNetworkQosPolicy(object): - """Fake one or more QoS policies.""" +class FakeNetworkQosRule(object): + """Fake one or more Network QoS rules.""" @staticmethod - def create_one_qos_policy(attrs=None): - """Create a fake QoS policy. + def create_one_qos_rule(attrs=None): + """Create a fake Network QoS rule. :param Dictionary attrs: A dictionary with all attributes @@ -746,71 +767,114 @@ class FakeNetworkQosPolicy(object): A FakeResource object with name, id, etc. """ attrs = attrs or {} - qos_id = attrs.get('id') or 'qos-policy-id-' + uuid.uuid4().hex - rule_attrs = {'qos_policy_id': qos_id} - rules = [ - FakeNetworkQosBandwidthLimitRule. - create_one_qos_bandwidth_limit_rule(rule_attrs)] # Set default attributes. - qos_policy_attrs = { - 'name': 'qos-policy-name-' + uuid.uuid4().hex, - 'id': qos_id, + type = attrs.get('type') or choice(VALID_QOS_RULES) + qos_rule_attrs = { + 'id': 'qos-rule-id-' + uuid.uuid4().hex, + 'qos_policy_id': 'qos-policy-id-' + uuid.uuid4().hex, 'tenant_id': 'project-id-' + uuid.uuid4().hex, - 'shared': False, - 'description': 'qos-policy-description-' + uuid.uuid4().hex, - 'rules': rules, + 'type': type, } + if type == RULE_TYPE_BANDWIDTH_LIMIT: + qos_rule_attrs['max_kbps'] = randint(1, 10000) + qos_rule_attrs['max_burst_kbits'] = randint(1, 10000) + elif type == RULE_TYPE_DSCP_MARKING: + qos_rule_attrs['dscp_mark'] = choice(VALID_DSCP_MARKS) + elif type == RULE_TYPE_MINIMUM_BANDWIDTH: + qos_rule_attrs['min_kbps'] = randint(1, 10000) + qos_rule_attrs['direction'] = 'egress' # Overwrite default attributes. - qos_policy_attrs.update(attrs) + qos_rule_attrs.update(attrs) - qos_policy = fakes.FakeResource( - info=copy.deepcopy(qos_policy_attrs), - loaded=True) + qos_rule = fakes.FakeResource(info=copy.deepcopy(qos_rule_attrs), + loaded=True) # Set attributes with special mapping in OpenStack SDK. - qos_policy.is_shared = qos_policy_attrs['shared'] - qos_policy.project_id = qos_policy_attrs['tenant_id'] + qos_rule.project_id = qos_rule['tenant_id'] - return qos_policy + return qos_rule @staticmethod - def create_qos_policies(attrs=None, count=2): - """Create multiple fake QoS policies. + def create_qos_rules(attrs=None, count=2): + """Create multiple fake Network QoS rules. :param Dictionary attrs: A dictionary with all attributes :param int count: - The number of QoS policies to fake + The number of Network QoS rule to fake :return: - A list of FakeResource objects faking the QoS policies + A list of FakeResource objects faking the Network QoS rules """ - qos_policies = [] + qos_rules = [] for i in range(0, count): - qos_policies.append( - FakeNetworkQosPolicy.create_one_qos_policy(attrs)) - - return qos_policies + qos_rules.append(FakeNetworkQosRule.create_one_qos_rule(attrs)) + return qos_rules @staticmethod - def get_qos_policies(qos_policies=None, count=2): - """Get an iterable MagicMock object with a list of faked QoS policies. + def get_qos_rules(qos_rules=None, count=2): + """Get a list of faked Network QoS rules. - If qos policies list is provided, then initialize the Mock object - with the list. Otherwise create one. + If Network QoS rules list is provided, then initialize the Mock + object with the list. Otherwise create one. :param List address scopes: - A list of FakeResource objects faking qos policies + A list of FakeResource objects faking Network QoS rules :param int count: - The number of QoS policies to fake + The number of QoS minimum bandwidth rules to fake :return: An iterable Mock object with side_effect set to a list of faked - QoS policies + qos minimum bandwidth rules """ - if qos_policies is None: - qos_policies = FakeNetworkQosPolicy.create_qos_policies(count) - return mock.Mock(side_effect=qos_policies) + if qos_rules is None: + qos_rules = (FakeNetworkQosRule.create_qos_rules(count)) + return mock.Mock(side_effect=qos_rules) + + +class FakeNetworkQosRuleType(object): + """Fake one or more Network QoS rule types.""" + + @staticmethod + def create_one_qos_rule_type(attrs=None): + """Create a fake Network QoS rule type. + + :param Dictionary attrs: + A dictionary with all attributes + :return: + A FakeResource object with name, id, etc. + """ + attrs = attrs or {} + + # Set default attributes. + qos_rule_type_attrs = { + 'type': 'rule-type-' + uuid.uuid4().hex, + } + + # Overwrite default attributes. + qos_rule_type_attrs.update(attrs) + + return fakes.FakeResource( + info=copy.deepcopy(qos_rule_type_attrs), + loaded=True) + + @staticmethod + def create_qos_rule_types(attrs=None, count=2): + """Create multiple fake Network QoS rule types. + + :param Dictionary attrs: + A dictionary with all attributes + :param int count: + The number of QoS rule types to fake + :return: + A list of FakeResource objects faking the QoS rule types + """ + qos_rule_types = [] + for i in range(0, count): + qos_rule_types.append( + FakeNetworkQosRuleType.create_one_qos_rule_type(attrs)) + + return qos_rule_types class FakeRouter(object): @@ -1213,6 +1277,51 @@ class FakeFloatingIP(object): return mock.Mock(side_effect=floating_ips) +class FakeNetworkMeter(object): + """Fake network meter""" + + @staticmethod + def create_one_meter(attrs=None): + """Create metering pool""" + attrs = attrs or {} + + meter_attrs = { + 'id': 'meter-id-' + uuid.uuid4().hex, + 'name': 'meter-name-' + uuid.uuid4().hex, + 'description': 'meter-description-' + uuid.uuid4().hex, + 'tenant_id': 'project-id-' + uuid.uuid4().hex, + 'shared': False + } + + meter_attrs.update(attrs) + + meter = fakes.FakeResource( + info=copy.deepcopy(meter_attrs), + loaded=True) + + meter.project_id = meter_attrs['tenant_id'] + + return meter + + @staticmethod + def create_meter(attrs=None, count=2): + """Create multiple meters""" + + meters = [] + for i in range(0, count): + meters.append(FakeNetworkMeter. + create_one_meter(attrs)) + return meters + + @staticmethod + def get_meter(meter=None, count=2): + """Get a list of meters""" + if meter is None: + meter = (FakeNetworkMeter. + create_meter(count)) + return mock.Mock(side_effect=meter) + + class FakeSubnetPool(object): """Fake one or more subnet pools.""" diff --git a/openstackclient/tests/unit/network/v2/test_floating_ip.py b/openstackclient/tests/unit/network/v2/test_floating_ip.py index 63d22bf8..e395300d 100644 --- a/openstackclient/tests/unit/network/v2/test_floating_ip.py +++ b/openstackclient/tests/unit/network/v2/test_floating_ip.py @@ -208,13 +208,19 @@ class TestDeleteFloatingIPNetwork(TestFloatingIPNetwork): super(TestDeleteFloatingIPNetwork, self).setUp() self.network.delete_ip = mock.Mock(return_value=None) - self.network.find_ip = ( - network_fakes.FakeFloatingIP.get_floating_ips(self.floating_ips)) # Get the command object to test self.cmd = floating_ip.DeleteFloatingIP(self.app, self.namespace) - def test_floating_ip_delete(self): + @mock.patch( + "openstackclient.tests.unit.network.v2.test_floating_ip." + + "floating_ip._find_floating_ip" + ) + def test_floating_ip_delete(self, find_floating_ip_mock): + find_floating_ip_mock.side_effect = [ + (self.floating_ips[0], []), + (self.floating_ips[1], []), + ] arglist = [ self.floating_ips[0].id, ] @@ -225,12 +231,24 @@ class TestDeleteFloatingIPNetwork(TestFloatingIPNetwork): result = self.cmd.take_action(parsed_args) - self.network.find_ip.assert_called_once_with( - self.floating_ips[0].id, ignore_missing=False) + find_floating_ip_mock.assert_called_once_with( + mock.ANY, + [], + self.floating_ips[0].id, + ignore_missing=False, + ) self.network.delete_ip.assert_called_once_with(self.floating_ips[0]) self.assertIsNone(result) - def test_multi_floating_ips_delete(self): + @mock.patch( + "openstackclient.tests.unit.network.v2.test_floating_ip." + + "floating_ip._find_floating_ip" + ) + def test_floating_ip_delete_multi(self, find_floating_ip_mock): + find_floating_ip_mock.side_effect = [ + (self.floating_ips[0], []), + (self.floating_ips[1], []), + ] arglist = [] verifylist = [] @@ -243,13 +261,37 @@ class TestDeleteFloatingIPNetwork(TestFloatingIPNetwork): result = self.cmd.take_action(parsed_args) + calls = [ + call( + mock.ANY, + [], + self.floating_ips[0].id, + ignore_missing=False, + ), + call( + mock.ANY, + [], + self.floating_ips[1].id, + ignore_missing=False, + ), + ] + find_floating_ip_mock.assert_has_calls(calls) + calls = [] for f in self.floating_ips: calls.append(call(f)) self.network.delete_ip.assert_has_calls(calls) self.assertIsNone(result) - def test_multi_floating_ips_delete_with_exception(self): + @mock.patch( + "openstackclient.tests.unit.network.v2.test_floating_ip." + + "floating_ip._find_floating_ip" + ) + def test_floating_ip_delete_multi_exception(self, find_floating_ip_mock): + find_floating_ip_mock.side_effect = [ + (self.floating_ips[0], []), + exceptions.CommandError, + ] arglist = [ self.floating_ips[0].id, 'unexist_floating_ip', @@ -260,21 +302,24 @@ class TestDeleteFloatingIPNetwork(TestFloatingIPNetwork): ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) - find_mock_result = [self.floating_ips[0], exceptions.CommandError] - self.network.find_ip = ( - mock.Mock(side_effect=find_mock_result) - ) - try: self.cmd.take_action(parsed_args) self.fail('CommandError should be raised.') except exceptions.CommandError as e: self.assertEqual('1 of 2 floating_ips failed to delete.', str(e)) - self.network.find_ip.assert_any_call( - self.floating_ips[0].id, ignore_missing=False) - self.network.find_ip.assert_any_call( - 'unexist_floating_ip', ignore_missing=False) + find_floating_ip_mock.assert_any_call( + mock.ANY, + [], + self.floating_ips[0].id, + ignore_missing=False, + ) + find_floating_ip_mock.assert_any_call( + mock.ANY, + [], + 'unexist_floating_ip', + ignore_missing=False, + ) self.network.delete_ip.assert_called_once_with( self.floating_ips[0] ) @@ -534,7 +579,12 @@ class TestShowFloatingIPNetwork(TestFloatingIPNetwork): # Get the command object to test self.cmd = floating_ip.ShowFloatingIP(self.app, self.namespace) - def test_floating_ip_show(self): + @mock.patch( + "openstackclient.tests.unit.network.v2.test_floating_ip." + + "floating_ip._find_floating_ip" + ) + def test_floating_ip_show(self, find_floating_ip_mock): + find_floating_ip_mock.return_value = (self.floating_ip, []) arglist = [ self.floating_ip.id, ] @@ -545,9 +595,11 @@ class TestShowFloatingIPNetwork(TestFloatingIPNetwork): columns, data = self.cmd.take_action(parsed_args) - self.network.find_ip.assert_called_once_with( + find_floating_ip_mock.assert_called_once_with( + mock.ANY, + [], self.floating_ip.id, - ignore_missing=False + ignore_missing=False, ) self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/network/v2/test_ip_availability.py b/openstackclient/tests/unit/network/v2/test_ip_availability.py index 4bdbddc4..c7c5a9b4 100644 --- a/openstackclient/tests/unit/network/v2/test_ip_availability.py +++ b/openstackclient/tests/unit/network/v2/test_ip_availability.py @@ -118,8 +118,10 @@ class TestListIPAvailability(TestIPAvailability): class TestShowIPAvailability(TestIPAvailability): + _network = network_fakes.FakeNetwork.create_one_network() _ip_availability = \ - network_fakes.FakeIPAvailability.create_one_ip_availability() + network_fakes.FakeIPAvailability.create_one_ip_availability( + attrs={'network_id': _network.id}) columns = ( 'network_id', @@ -144,6 +146,8 @@ class TestShowIPAvailability(TestIPAvailability): self.network.find_network_ip_availability = mock.Mock( return_value=self._ip_availability) + self.network.find_network = mock.Mock( + return_value=self._network) # Get the command object to test self.cmd = ip_availability.ShowIPAvailability( @@ -166,8 +170,10 @@ class TestShowIPAvailability(TestIPAvailability): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) self.network.find_network_ip_availability.assert_called_once_with( + self._ip_availability.network_id, + ignore_missing=False) + self.network.find_network.assert_called_once_with( self._ip_availability.network_name, ignore_missing=False) - self.assertEqual(self.columns, columns) self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/network/v2/test_meter.py b/openstackclient/tests/unit/network/v2/test_meter.py new file mode 100644 index 00000000..b393f7fa --- /dev/null +++ b/openstackclient/tests/unit/network/v2/test_meter.py @@ -0,0 +1,304 @@ +# 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 mock +from mock import call + +from osc_lib import exceptions + +from openstackclient.network.v2 import meter +from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes_v3 +from openstackclient.tests.unit.network.v2 import fakes as network_fakes +from openstackclient.tests.unit import utils as tests_utils + + +class TestMeter(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestMeter, self).setUp() + self.network = self.app.client_manager.network + self.projects_mock = self.app.client_manager.identity.projects + self.domains_mock = self.app.client_manager.identity.domains + + +class TestCreateMeter(TestMeter): + project = identity_fakes_v3.FakeProject.create_one_project() + domain = identity_fakes_v3.FakeDomain.create_one_domain() + + new_meter = ( + network_fakes.FakeNetworkMeter. + create_one_meter() + ) + columns = ( + 'description', + 'id', + 'name', + 'project_id', + 'shared', + ) + + data = ( + new_meter.description, + new_meter.id, + new_meter.name, + new_meter.project_id, + new_meter.shared, + ) + + def setUp(self): + super(TestCreateMeter, self).setUp() + self.network.create_metering_label = mock.Mock( + return_value=self.new_meter) + self.projects_mock.get.return_value = self.project + self.cmd = meter.CreateMeter(self.app, self.namespace) + + def test_create_no_options(self): + arglist = [] + verifylist = [] + + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_create_default_options(self): + arglist = [ + self.new_meter.name, + ] + + verifylist = [ + ('name', self.new_meter.name), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_metering_label.assert_called_once_with( + **{'name': self.new_meter.name} + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_create_all_options(self): + arglist = [ + "--description", self.new_meter.description, + "--project", self.new_meter.project_id, + "--project-domain", self.domain.name, + "--share", + self.new_meter.name, + ] + + verifylist = [ + ('description', self.new_meter.description), + ('name', self.new_meter.name), + ('project', self.new_meter.project_id), + ('project_domain', self.domain.name), + ('share', True), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_metering_label.assert_called_once_with( + **{'description': self.new_meter.description, + 'name': self.new_meter.name, + 'tenant_id': self.project.id, + 'shared': True, } + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + +class TestDeleteMeter(TestMeter): + + def setUp(self): + super(TestDeleteMeter, self).setUp() + + self.meter_list = \ + network_fakes.FakeNetworkMeter.create_meter(count=2) + + self.network.delete_metering_label = mock.Mock(return_value=None) + + self.network.find_metering_label = network_fakes \ + .FakeNetworkMeter.get_meter( + meter=self.meter_list + ) + + self.cmd = meter.DeleteMeter(self.app, self.namespace) + + def test_delete_one_meter(self): + arglist = [ + self.meter_list[0].name, + ] + verifylist = [ + ('meter', [self.meter_list[0].name]), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + self.network.delete_metering_label.assert_called_once_with( + self.meter_list[0] + ) + self.assertIsNone(result) + + def test_delete_multiple_meters(self): + arglist = [] + for n in self.meter_list: + arglist.append(n.id) + verifylist = [ + ('meter', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + calls = [] + for n in self.meter_list: + calls.append(call(n)) + self.network.delete_metering_label.assert_has_calls(calls) + self.assertIsNone(result) + + def test_delete_multiple_meter_exception(self): + arglist = [ + self.meter_list[0].id, + 'xxxx-yyyy-zzzz', + self.meter_list[1].id, + ] + verifylist = [ + ('meter', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + return_find = [ + self.meter_list[0], + exceptions.NotFound('404'), + self.meter_list[1], + ] + self.network.find_meter = mock.Mock(side_effect=return_find) + + ret_delete = [ + None, + exceptions.NotFound('404'), + ] + self.network.delete_metering_label = mock.Mock(side_effect=ret_delete) + + self.assertRaises(exceptions.CommandError, self.cmd.take_action, + parsed_args) + + calls = [ + call(self.meter_list[0]), + call(self.meter_list[1]), + ] + self.network.delete_metering_label.assert_has_calls(calls) + + +class TestListMeter(TestMeter): + + meter_list = \ + network_fakes.FakeNetworkMeter.create_meter(count=2) + + columns = ( + 'ID', + 'Name', + 'Description', + 'Shared', + ) + + data = [] + + for meters in meter_list: + data.append(( + meters.id, + meters.name, + meters.description, + meters.shared, + )) + + def setUp(self): + super(TestListMeter, self).setUp() + + self.network.metering_labels = mock.Mock( + return_value=self.meter_list + ) + + self.cmd = meter.ListMeter(self.app, self.namespace) + + def test_meter_list(self): + arglist = [] + verifylist = [] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.metering_labels.assert_called_with() + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + +class TestShowMeter(TestMeter): + new_meter = ( + network_fakes.FakeNetworkMeter. + create_one_meter() + ) + columns = ( + 'description', + 'id', + 'name', + 'project_id', + 'shared', + ) + + data = ( + new_meter.description, + new_meter.id, + new_meter.name, + new_meter.project_id, + new_meter.shared, + ) + + def setUp(self): + super(TestShowMeter, self).setUp() + + self.cmd = meter.ShowMeter(self.app, self.namespace) + + self.network.find_metering_label = \ + mock.Mock(return_value=self.new_meter) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_meter_show_option(self): + arglist = [ + self.new_meter.name, + ] + verifylist = [ + ('meter', self.new_meter.name), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.find_metering_label.assert_called_with( + self.new_meter.name, ignore_missing=False + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) diff --git a/openstackclient/tests/unit/network/v2/test_network_agent.py b/openstackclient/tests/unit/network/v2/test_network_agent.py index 9f5b442a..2fc0c043 100644 --- a/openstackclient/tests/unit/network/v2/test_network_agent.py +++ b/openstackclient/tests/unit/network/v2/test_network_agent.py @@ -130,6 +130,7 @@ class TestListNetworkAgent(TestNetworkAgent): ) data = [] for agent in network_agents: + agent.agent_type = 'DHCP agent' data.append(( agent.id, agent.agent_type, @@ -159,7 +160,43 @@ class TestListNetworkAgent(TestNetworkAgent): self.assertEqual(self.columns, columns) self.assertEqual(self.data, list(data)) + def test_network_agents_list_agent_type(self): + arglist = [ + '--agent-type', 'dhcp', + ] + verifylist = [ + ('agent_type', 'dhcp'), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.agents.assert_called_once_with(**{ + 'agent_type': self.network_agents[0].agent_type, + }) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_network_agents_list_host(self): + arglist = [ + '--host', self.network_agents[0].host, + ] + verifylist = [ + ('host', self.network_agents[0].host), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.agents.assert_called_once_with(**{ + 'host': self.network_agents[0].host, + }) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + +# TODO(huanxuan): Also update by the new attribute name +# "is_admin_state_up" after sdk 0.9.12 class TestSetNetworkAgent(TestNetworkAgent): _network_agent = ( @@ -289,6 +326,6 @@ class TestShowNetworkAgent(TestNetworkAgent): columns, data = self.cmd.take_action(parsed_args) self.network.get_agent.assert_called_once_with( - self._network_agent.id, ignore_missing=False) + self._network_agent.id) self.assertEqual(self.columns, columns) self.assertEqual(list(self.data), list(data)) diff --git a/openstackclient/tests/unit/network/v2/test_network_qos_rule.py b/openstackclient/tests/unit/network/v2/test_network_qos_rule.py new file mode 100644 index 00000000..41ccae32 --- /dev/null +++ b/openstackclient/tests/unit/network/v2/test_network_qos_rule.py @@ -0,0 +1,1049 @@ +# 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 mock + +from osc_lib import exceptions + +from openstackclient.network.v2 import network_qos_rule +from openstackclient.tests.unit.network.v2 import fakes as network_fakes +from openstackclient.tests.unit import utils as tests_utils + + +RULE_TYPE_BANDWIDTH_LIMIT = 'bandwidth-limit' +RULE_TYPE_DSCP_MARKING = 'dscp-marking' +RULE_TYPE_MINIMUM_BANDWIDTH = 'minimum-bandwidth' +DSCP_VALID_MARKS = [0, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, + 34, 36, 38, 40, 46, 48, 56] + + +class TestNetworkQosRule(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestNetworkQosRule, self).setUp() + # Get a shortcut to the network client + self.network = self.app.client_manager.network + self.qos_policy = (network_fakes.FakeNetworkQosPolicy. + create_one_qos_policy()) + self.network.find_qos_policy = mock.Mock(return_value=self.qos_policy) + + +class TestCreateNetworkQosRuleMinimumBandwidth(TestNetworkQosRule): + + def test_check_type_parameters(self): + pass + + def setUp(self): + super(TestCreateNetworkQosRuleMinimumBandwidth, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_MINIMUM_BANDWIDTH} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.columns = ( + 'direction', + 'id', + 'min_kbps', + 'project_id', + 'qos_policy_id', + 'type' + ) + + self.data = ( + self.new_rule.direction, + self.new_rule.id, + self.new_rule.min_kbps, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + self.network.create_qos_minimum_bandwidth_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.CreateNetworkQosRule(self.app, + self.namespace) + + def test_create_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_create_default_options(self): + arglist = [ + '--type', RULE_TYPE_MINIMUM_BANDWIDTH, + '--min-kbps', str(self.new_rule.min_kbps), + '--egress', + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_MINIMUM_BANDWIDTH), + ('min_kbps', self.new_rule.min_kbps), + ('egress', True), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_qos_minimum_bandwidth_rule.assert_called_once_with( + self.qos_policy.id, + **{'min_kbps': self.new_rule.min_kbps, + 'direction': self.new_rule.direction} + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_create_wrong_options(self): + arglist = [ + '--type', RULE_TYPE_MINIMUM_BANDWIDTH, + '--max-kbps', '10000', + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_MINIMUM_BANDWIDTH), + ('max_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('"Create" rule command for type "minimum-bandwidth" ' + 'requires arguments min_kbps, direction') + self.assertEqual(msg, str(e)) + + +class TestCreateNetworkQosRuleDSCPMarking(TestNetworkQosRule): + + def test_check_type_parameters(self): + pass + + def setUp(self): + super(TestCreateNetworkQosRuleDSCPMarking, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_DSCP_MARKING} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.columns = ( + 'dscp_mark', + 'id', + 'project_id', + 'qos_policy_id', + 'type' + ) + + self.data = ( + self.new_rule.dscp_mark, + self.new_rule.id, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + self.network.create_qos_dscp_marking_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.CreateNetworkQosRule(self.app, + self.namespace) + + def test_create_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_create_default_options(self): + arglist = [ + '--type', RULE_TYPE_DSCP_MARKING, + '--dscp-mark', str(self.new_rule.dscp_mark), + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_DSCP_MARKING), + ('dscp_mark', self.new_rule.dscp_mark), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.create_qos_dscp_marking_rule.assert_called_once_with( + self.qos_policy.id, + **{'dscp_mark': self.new_rule.dscp_mark} + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_create_wrong_options(self): + arglist = [ + '--type', RULE_TYPE_DSCP_MARKING, + '--max-kbps', '10000', + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_DSCP_MARKING), + ('max_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('"Create" rule command for type "dscp-marking" ' + 'requires arguments dscp_mark') + self.assertEqual(msg, str(e)) + + +class TestCreateNetworkQosRuleBandwidtLimit(TestNetworkQosRule): + + def test_check_type_parameters(self): + pass + + def setUp(self): + super(TestCreateNetworkQosRuleBandwidtLimit, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_BANDWIDTH_LIMIT} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.columns = ( + 'id', + 'max_burst_kbits', + 'max_kbps', + 'project_id', + 'qos_policy_id', + 'type' + ) + + self.data = ( + self.new_rule.id, + self.new_rule.max_burst_kbits, + self.new_rule.max_kbps, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + self.network.create_qos_bandwidth_limit_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.CreateNetworkQosRule(self.app, + self.namespace) + + def test_create_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_create_default_options(self): + arglist = [ + '--type', RULE_TYPE_BANDWIDTH_LIMIT, + '--max-kbps', str(self.new_rule.max_kbps), + '--max-burst-kbits', str(self.new_rule.max_burst_kbits), + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_BANDWIDTH_LIMIT), + ('max_kbps', self.new_rule.max_kbps), + ('max_burst_kbits', self.new_rule.max_burst_kbits), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = (self.cmd.take_action(parsed_args)) + + self.network.create_qos_bandwidth_limit_rule.assert_called_once_with( + self.qos_policy.id, + **{'max_kbps': self.new_rule.max_kbps, + 'max_burst_kbps': self.new_rule.max_burst_kbits} + ) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, data) + + def test_create_wrong_options(self): + arglist = [ + '--type', RULE_TYPE_BANDWIDTH_LIMIT, + '--min-kbps', '10000', + self.new_rule.qos_policy_id, + ] + + verifylist = [ + ('type', RULE_TYPE_BANDWIDTH_LIMIT), + ('min_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('"Create" rule command for type "bandwidth-limit" ' + 'requires arguments max_kbps, max_burst_kbps') + self.assertEqual(msg, str(e)) + + +class TestDeleteNetworkQosRuleMinimumBandwidth(TestNetworkQosRule): + + def setUp(self): + super(TestDeleteNetworkQosRuleMinimumBandwidth, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_MINIMUM_BANDWIDTH} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.network.delete_qos_minimum_bandwidth_rule = mock.Mock( + return_value=None) + self.network.find_qos_minimum_bandwidth_rule = ( + network_fakes.FakeNetworkQosRule.get_qos_rules( + qos_rules=self.new_rule) + ) + + # Get the command object to test + self.cmd = network_qos_rule.DeleteNetworkQosRule(self.app, + self.namespace) + + def test_qos_policy_delete(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.network.find_qos_policy.assert_called_once_with( + self.qos_policy.id, ignore_missing=False) + self.network.delete_qos_minimum_bandwidth_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertIsNone(result) + + def test_qos_policy_delete_error(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + self.network.delete_qos_minimum_bandwidth_rule.side_effect = \ + Exception('Error message') + try: + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to delete Network QoS rule ID "%(rule)s": %(e)s' % + {'rule': self.new_rule.id, 'e': 'Error message'}) + self.assertEqual(msg, str(e)) + + +class TestDeleteNetworkQosRuleDSCPMarking(TestNetworkQosRule): + + def setUp(self): + super(TestDeleteNetworkQosRuleDSCPMarking, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_DSCP_MARKING} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.network.delete_qos_dscp_marking_rule = mock.Mock( + return_value=None) + self.network.find_qos_dscp_marking_rule = ( + network_fakes.FakeNetworkQosRule.get_qos_rules( + qos_rules=self.new_rule) + ) + + # Get the command object to test + self.cmd = network_qos_rule.DeleteNetworkQosRule(self.app, + self.namespace) + + def test_qos_policy_delete(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.network.find_qos_policy.assert_called_once_with( + self.qos_policy.id, ignore_missing=False) + self.network.delete_qos_dscp_marking_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertIsNone(result) + + def test_qos_policy_delete_error(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + self.network.delete_qos_dscp_marking_rule.side_effect = \ + Exception('Error message') + try: + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to delete Network QoS rule ID "%(rule)s": %(e)s' % + {'rule': self.new_rule.id, 'e': 'Error message'}) + self.assertEqual(msg, str(e)) + + +class TestDeleteNetworkQosRuleBandwidthLimit(TestNetworkQosRule): + + def setUp(self): + super(TestDeleteNetworkQosRuleBandwidthLimit, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_BANDWIDTH_LIMIT} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.network.delete_qos_bandwidth_limit_rule = mock.Mock( + return_value=None) + self.network.find_qos_bandwidth_limit_rule = ( + network_fakes.FakeNetworkQosRule.get_qos_rules( + qos_rules=self.new_rule) + ) + + # Get the command object to test + self.cmd = network_qos_rule.DeleteNetworkQosRule(self.app, + self.namespace) + + def test_qos_policy_delete(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + self.network.find_qos_policy.assert_called_once_with( + self.qos_policy.id, ignore_missing=False) + self.network.delete_qos_bandwidth_limit_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertIsNone(result) + + def test_qos_policy_delete_error(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + self.network.delete_qos_bandwidth_limit_rule.side_effect = \ + Exception('Error message') + try: + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to delete Network QoS rule ID "%(rule)s": %(e)s' % + {'rule': self.new_rule.id, 'e': 'Error message'}) + self.assertEqual(msg, str(e)) + + +class TestSetNetworkQosRuleMinimumBandwidth(TestNetworkQosRule): + + def setUp(self): + super(TestSetNetworkQosRuleMinimumBandwidth, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_MINIMUM_BANDWIDTH} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs=attrs) + self.qos_policy.rules = [self.new_rule] + self.network.update_qos_minimum_bandwidth_rule = mock.Mock( + return_value=None) + self.network.find_qos_minimum_bandwidth_rule = mock.Mock( + return_value=self.new_rule) + self.network.find_qos_policy = mock.Mock( + return_value=self.qos_policy) + + # Get the command object to test + self.cmd = (network_qos_rule.SetNetworkQosRule(self.app, + self.namespace)) + + def test_set_nothing(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + self.network.update_qos_minimum_bandwidth_rule.assert_called_with( + self.new_rule, self.qos_policy.id) + self.assertIsNone(result) + + def test_set_min_kbps(self): + self._set_min_kbps() + + def test_set_min_kbps_to_zero(self): + self._set_min_kbps(min_kbps=0) + + def _set_min_kbps(self, min_kbps=None): + if min_kbps: + previous_min_kbps = self.new_rule.min_kbps + self.new_rule.min_kbps = min_kbps + + arglist = [ + '--min-kbps', str(self.new_rule.min_kbps), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('min_kbps', self.new_rule.min_kbps), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'min_kbps': self.new_rule.min_kbps, + } + self.network.update_qos_minimum_bandwidth_rule.assert_called_with( + self.new_rule, self.qos_policy.id, **attrs) + self.assertIsNone(result) + + if min_kbps: + self.new_rule.min_kbps = previous_min_kbps + + def test_set_wrong_options(self): + arglist = [ + '--max-kbps', str(10000), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('max_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type ' + '"minimum-bandwidth" only requires arguments min_kbps, ' + 'direction' % {'rule': self.new_rule.id}) + self.assertEqual(msg, str(e)) + + +class TestSetNetworkQosRuleDSCPMarking(TestNetworkQosRule): + + def setUp(self): + super(TestSetNetworkQosRuleDSCPMarking, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_DSCP_MARKING} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs=attrs) + self.qos_policy.rules = [self.new_rule] + self.network.update_qos_dscp_marking_rule = mock.Mock( + return_value=None) + self.network.find_qos_dscp_marking_rule = mock.Mock( + return_value=self.new_rule) + self.network.find_qos_policy = mock.Mock( + return_value=self.qos_policy) + + # Get the command object to test + self.cmd = (network_qos_rule.SetNetworkQosRule(self.app, + self.namespace)) + + def test_set_nothing(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + self.network.update_qos_dscp_marking_rule.assert_called_with( + self.new_rule, self.qos_policy.id) + self.assertIsNone(result) + + def test_set_dscp_mark(self): + self._set_dscp_mark() + + def test_set_dscp_mark_to_zero(self): + self._set_dscp_mark(dscp_mark=0) + + def _set_dscp_mark(self, dscp_mark=None): + if dscp_mark: + previous_dscp_mark = self.new_rule.dscp_mark + self.new_rule.dscp_mark = dscp_mark + + arglist = [ + '--dscp-mark', str(self.new_rule.dscp_mark), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('dscp_mark', self.new_rule.dscp_mark), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'dscp_mark': self.new_rule.dscp_mark, + } + self.network.update_qos_dscp_marking_rule.assert_called_with( + self.new_rule, self.qos_policy.id, **attrs) + self.assertIsNone(result) + + if dscp_mark: + self.new_rule.dscp_mark = previous_dscp_mark + + def test_set_wrong_options(self): + arglist = [ + '--max-kbps', str(10000), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('max_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type ' + '"dscp-marking" only requires arguments dscp_mark' % + {'rule': self.new_rule.id}) + self.assertEqual(msg, str(e)) + + +class TestSetNetworkQosRuleBandwidthLimit(TestNetworkQosRule): + + def setUp(self): + super(TestSetNetworkQosRuleBandwidthLimit, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_BANDWIDTH_LIMIT} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs=attrs) + self.qos_policy.rules = [self.new_rule] + self.network.update_qos_bandwidth_limit_rule = mock.Mock( + return_value=None) + self.network.find_qos_bandwidth_limit_rule = mock.Mock( + return_value=self.new_rule) + self.network.find_qos_policy = mock.Mock( + return_value=self.qos_policy) + + # Get the command object to test + self.cmd = (network_qos_rule.SetNetworkQosRule(self.app, + self.namespace)) + + def test_set_nothing(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + self.network.update_qos_bandwidth_limit_rule.assert_called_with( + self.new_rule, self.qos_policy.id) + self.assertIsNone(result) + + def test_set_max_kbps(self): + self._set_max_kbps() + + def test_set_max_kbps_to_zero(self): + self._set_max_kbps(max_kbps=0) + + def _set_max_kbps(self, max_kbps=None): + if max_kbps: + previous_max_kbps = self.new_rule.max_kbps + self.new_rule.max_kbps = max_kbps + + arglist = [ + '--max-kbps', str(self.new_rule.max_kbps), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('max_kbps', self.new_rule.max_kbps), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'max_kbps': self.new_rule.max_kbps, + } + self.network.update_qos_bandwidth_limit_rule.assert_called_with( + self.new_rule, self.qos_policy.id, **attrs) + self.assertIsNone(result) + + if max_kbps: + self.new_rule.max_kbps = previous_max_kbps + + def test_set_max_burst_kbits(self): + self._set_max_burst_kbits() + + def test_set_max_burst_kbits_to_zero(self): + self._set_max_burst_kbits(max_burst_kbits=0) + + def _set_max_burst_kbits(self, max_burst_kbits=None): + if max_burst_kbits: + previous_max_burst_kbits = self.new_rule.max_burst_kbits + self.new_rule.max_burst_kbits = max_burst_kbits + + arglist = [ + '--max-burst-kbits', str(self.new_rule.max_burst_kbits), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('max_burst_kbits', self.new_rule.max_burst_kbits), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + attrs = { + 'max_burst_kbps': self.new_rule.max_burst_kbits, + } + self.network.update_qos_bandwidth_limit_rule.assert_called_with( + self.new_rule, self.qos_policy.id, **attrs) + self.assertIsNone(result) + + if max_burst_kbits: + self.new_rule.max_burst_kbits = previous_max_burst_kbits + + def test_set_wrong_options(self): + arglist = [ + '--min-kbps', str(10000), + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('min_kbps', 10000), + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + try: + self.cmd.take_action(parsed_args) + except exceptions.CommandError as e: + msg = ('Failed to set Network QoS rule ID "%(rule)s": Rule type ' + '"bandwidth-limit" only requires arguments max_kbps, ' + 'max_burst_kbps' % {'rule': self.new_rule.id}) + self.assertEqual(msg, str(e)) + + +class TestListNetworkQosRule(TestNetworkQosRule): + + def setUp(self): + super(TestListNetworkQosRule, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_MINIMUM_BANDWIDTH} + self.new_rule_min_bw = (network_fakes.FakeNetworkQosRule. + create_one_qos_rule(attrs=attrs)) + attrs['type'] = RULE_TYPE_DSCP_MARKING + self.new_rule_dscp_mark = (network_fakes.FakeNetworkQosRule. + create_one_qos_rule(attrs=attrs)) + attrs['type'] = RULE_TYPE_BANDWIDTH_LIMIT + self.new_rule_max_bw = (network_fakes.FakeNetworkQosRule. + create_one_qos_rule(attrs=attrs)) + self.qos_policy.rules = [self.new_rule_min_bw, + self.new_rule_dscp_mark, + self.new_rule_max_bw] + self.network.find_qos_minimum_bandwidth_rule = mock.Mock( + return_value=self.new_rule_min_bw) + self.network.find_qos_dscp_marking_rule = mock.Mock( + return_value=self.new_rule_dscp_mark) + self.network.find_qos_bandwidth_limit_rule = mock.Mock( + return_value=self.new_rule_max_bw) + self.columns = ( + 'ID', + 'QoS Policy ID', + 'Type', + 'Max Kbps', + 'Max Burst Kbits', + 'Min Kbps', + 'DSCP mark', + 'Direction', + ) + self.data = [] + for index in range(len(self.qos_policy.rules)): + self.data.append(( + self.qos_policy.rules[index].id, + self.qos_policy.rules[index].qos_policy_id, + self.qos_policy.rules[index].type, + getattr(self.qos_policy.rules[index], 'max_kbps', ''), + getattr(self.qos_policy.rules[index], 'max_burst_kbps', ''), + getattr(self.qos_policy.rules[index], 'min_kbps', ''), + getattr(self.qos_policy.rules[index], 'dscp_mark', ''), + getattr(self.qos_policy.rules[index], 'direction', ''), + )) + # Get the command object to test + self.cmd = network_qos_rule.ListNetworkQosRule(self.app, + self.namespace) + + def test_qos_rule_list(self): + arglist = [ + self.qos_policy.id + ] + verifylist = [ + ('qos_policy', self.qos_policy.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.find_qos_policy.assert_called_once_with( + self.qos_policy.id, ignore_missing=False) + self.assertEqual(self.columns, columns) + list_data = list(data) + self.assertEqual(len(self.data), len(list_data)) + for index in range(len(list_data)): + self.assertEqual(self.data[index], list_data[index]) + + +class TestShowNetworkQosRuleMinimumBandwidth(TestNetworkQosRule): + + def setUp(self): + super(TestShowNetworkQosRuleMinimumBandwidth, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_MINIMUM_BANDWIDTH} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.columns = ( + 'direction', + 'id', + 'min_kbps', + 'project_id', + 'qos_policy_id', + 'type' + ) + self.data = ( + self.new_rule.direction, + self.new_rule.id, + self.new_rule.min_kbps, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + + self.network.get_qos_minimum_bandwidth_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.ShowNetworkQosRule(self.app, + self.namespace) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.get_qos_minimum_bandwidth_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertEqual(self.columns, columns) + self.assertEqual(list(self.data), list(data)) + + +class TestShowNetworkQosDSCPMarking(TestNetworkQosRule): + + def setUp(self): + super(TestShowNetworkQosDSCPMarking, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_DSCP_MARKING} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.columns = ( + 'dscp_mark', + 'id', + 'project_id', + 'qos_policy_id', + 'type' + ) + self.data = ( + self.new_rule.dscp_mark, + self.new_rule.id, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + + self.network.get_qos_dscp_marking_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.ShowNetworkQosRule(self.app, + self.namespace) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.get_qos_dscp_marking_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertEqual(self.columns, columns) + self.assertEqual(list(self.data), list(data)) + + +class TestShowNetworkQosBandwidthLimit(TestNetworkQosRule): + + def setUp(self): + super(TestShowNetworkQosBandwidthLimit, self).setUp() + attrs = {'qos_policy_id': self.qos_policy.id, + 'type': RULE_TYPE_BANDWIDTH_LIMIT} + self.new_rule = network_fakes.FakeNetworkQosRule.create_one_qos_rule( + attrs) + self.qos_policy.rules = [self.new_rule] + self.columns = ( + 'id', + 'max_burst_kbits', + 'max_kbps', + 'project_id', + 'qos_policy_id', + 'type' + ) + self.data = ( + self.new_rule.id, + self.new_rule.max_burst_kbits, + self.new_rule.max_kbps, + self.new_rule.project_id, + self.new_rule.qos_policy_id, + self.new_rule.type, + ) + + self.network.get_qos_bandwidth_limit_rule = mock.Mock( + return_value=self.new_rule) + + # Get the command object to test + self.cmd = network_qos_rule.ShowNetworkQosRule(self.app, + self.namespace) + + def test_show_no_options(self): + arglist = [] + verifylist = [] + + # Missing required args should bail here + self.assertRaises(tests_utils.ParserException, self.check_parser, + self.cmd, arglist, verifylist) + + def test_show_all_options(self): + arglist = [ + self.new_rule.qos_policy_id, + self.new_rule.id, + ] + verifylist = [ + ('qos_policy', self.new_rule.qos_policy_id), + ('id', self.new_rule.id), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.get_qos_bandwidth_limit_rule.assert_called_once_with( + self.new_rule.id, self.qos_policy.id) + self.assertEqual(self.columns, columns) + self.assertEqual(list(self.data), list(data)) diff --git a/openstackclient/tests/unit/network/v2/test_network_qos_rule_type.py b/openstackclient/tests/unit/network/v2/test_network_qos_rule_type.py new file mode 100644 index 00000000..b93abe80 --- /dev/null +++ b/openstackclient/tests/unit/network/v2/test_network_qos_rule_type.py @@ -0,0 +1,62 @@ +# 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 mock + +from openstackclient.network.v2 import network_qos_rule_type as _qos_rule_type +from openstackclient.tests.unit.network.v2 import fakes as network_fakes + + +class TestNetworkQosRuleType(network_fakes.TestNetworkV2): + + def setUp(self): + super(TestNetworkQosRuleType, self).setUp() + # Get a shortcut to the network client + self.network = self.app.client_manager.network + + +class TestListNetworkQosRuleType(TestNetworkQosRuleType): + + # The QoS policies to list up. + qos_rule_types = ( + network_fakes.FakeNetworkQosRuleType.create_qos_rule_types(count=3)) + columns = ( + 'Type', + ) + data = [] + for qos_rule_type in qos_rule_types: + data.append(( + qos_rule_type.type, + )) + + def setUp(self): + super(TestListNetworkQosRuleType, self).setUp() + self.network.qos_rule_types = mock.Mock( + return_value=self.qos_rule_types) + + # Get the command object to test + self.cmd = _qos_rule_type.ListNetworkQosRuleType(self.app, + self.namespace) + + def test_qos_rule_type_list(self): + arglist = [] + verifylist = [] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + columns, data = self.cmd.take_action(parsed_args) + + self.network.qos_rule_types.assert_called_once_with(**{}) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) diff --git a/openstackclient/tests/unit/network/v2/test_network_rbac.py b/openstackclient/tests/unit/network/v2/test_network_rbac.py index b884dbc0..935ce075 100644 --- a/openstackclient/tests/unit/network/v2/test_network_rbac.py +++ b/openstackclient/tests/unit/network/v2/test_network_rbac.py @@ -327,7 +327,12 @@ class TestListNetworkRABC(TestNetworkRBAC): 'Object Type', 'Object ID', ) - + columns_long = ( + 'ID', + 'Object Type', + 'Object ID', + 'Action', + ) data = [] for r in rbac_policies: data.append(( @@ -335,6 +340,14 @@ class TestListNetworkRABC(TestNetworkRBAC): r.object_type, r.object_id, )) + data_long = [] + for r in rbac_policies: + data_long.append(( + r.id, + r.object_type, + r.object_id, + r.action, + )) def setUp(self): super(TestListNetworkRABC, self).setUp() @@ -356,6 +369,55 @@ class TestListNetworkRABC(TestNetworkRBAC): self.assertEqual(self.columns, columns) self.assertEqual(self.data, list(data)) + def test_network_rbac_list_type_opt(self): + arglist = [ + '--type', self.rbac_policies[0].object_type, ] + verifylist = [ + ('type', self.rbac_policies[0].object_type)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # DisplayCommandBase.take_action() returns two tuples + columns, data = self.cmd.take_action(parsed_args) + + self.network.rbac_policies.assert_called_with(**{ + 'object_type': self.rbac_policies[0].object_type + }) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_network_rbac_list_action_opt(self): + arglist = [ + '--action', self.rbac_policies[0].action, ] + verifylist = [ + ('action', self.rbac_policies[0].action)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + # DisplayCommandBase.take_action() returns two tuples + columns, data = self.cmd.take_action(parsed_args) + + self.network.rbac_policies.assert_called_with(**{ + 'action': self.rbac_policies[0].action + }) + self.assertEqual(self.columns, columns) + self.assertEqual(self.data, list(data)) + + def test_network_rbac_list_with_long(self): + arglist = [ + '--long', + ] + + verifylist = [ + ('long', True), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + columns, data = self.cmd.take_action(parsed_args) + + self.network.rbac_policies.assert_called_with() + self.assertEqual(self.columns_long, columns) + self.assertEqual(self.data_long, list(data)) + class TestSetNetworkRBAC(TestNetworkRBAC): diff --git a/openstackclient/tests/unit/network/v2/test_router.py b/openstackclient/tests/unit/network/v2/test_router.py index 9183cb63..b837afd1 100644 --- a/openstackclient/tests/unit/network/v2/test_router.py +++ b/openstackclient/tests/unit/network/v2/test_router.py @@ -704,10 +704,10 @@ class TestSetRouter(TestRouter): parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - + routes = [{'destination': '10.20.30.0/24', + 'nexthop': '10.20.30.1'}] attrs = { - 'routes': self._router.routes + [{'destination': '10.20.30.0/24', - 'nexthop': '10.20.30.1'}], + 'routes': routes + self._router.routes } self.network.update_router.assert_called_once_with( self._router, **attrs) @@ -733,21 +733,31 @@ class TestSetRouter(TestRouter): self._router, **attrs) self.assertIsNone(result) - def test_set_route_no_route(self): + def test_set_route_overwrite_route(self): + _testrouter = network_fakes.FakeRouter.create_one_router( + {'routes': [{"destination": "10.0.0.2", + "nexthop": "1.1.1.1"}]}) + self.network.find_router = mock.Mock(return_value=_testrouter) arglist = [ - self._router.name, + _testrouter.name, '--route', 'destination=10.20.30.0/24,gateway=10.20.30.1', '--no-route', ] verifylist = [ - ('router', self._router.name), + ('router', _testrouter.name), ('routes', [{'destination': '10.20.30.0/24', 'gateway': '10.20.30.1'}]), ('no_route', True), ] - - self.assertRaises(tests_utils.ParserException, self.check_parser, - self.cmd, arglist, verifylist) + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + attrs = { + 'routes': [{'destination': '10.20.30.0/24', + 'nexthop': '10.20.30.1'}] + } + self.network.update_router.assert_called_once_with( + _testrouter, **attrs) + self.assertIsNone(result) def test_set_clear_routes(self): arglist = [ @@ -769,21 +779,31 @@ class TestSetRouter(TestRouter): self._router, **attrs) self.assertIsNone(result) - def test_set_route_clear_routes(self): + def test_overwrite_route_clear_routes(self): + _testrouter = network_fakes.FakeRouter.create_one_router( + {'routes': [{"destination": "10.0.0.2", + "nexthop": "1.1.1.1"}]}) + self.network.find_router = mock.Mock(return_value=_testrouter) arglist = [ - self._router.name, + _testrouter.name, '--route', 'destination=10.20.30.0/24,gateway=10.20.30.1', '--clear-routes', ] verifylist = [ - ('router', self._router.name), + ('router', _testrouter.name), ('routes', [{'destination': '10.20.30.0/24', 'gateway': '10.20.30.1'}]), ('clear_routes', True), ] - - self.assertRaises(tests_utils.ParserException, self.check_parser, - self.cmd, arglist, verifylist) + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + attrs = { + 'routes': [{'destination': '10.20.30.0/24', + 'nexthop': '10.20.30.1'}] + } + self.network.update_router.assert_called_once_with( + _testrouter, **attrs) + self.assertIsNone(result) def test_set_nothing(self): arglist = [ @@ -1021,3 +1041,16 @@ class TestUnsetRouter(TestRouter): parsed_args = self.check_parser(self.cmd, arglist, verifylist) self.assertRaises(exceptions.CommandError, self.cmd.take_action, parsed_args) + + def test_unset_router_external_gateway(self): + arglist = [ + '--external-gateway', + self._testrouter.name, + ] + verifylist = [('external_gateway', True)] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + attrs = {'external_gateway_info': {}} + self.network.update_router.assert_called_once_with( + self._testrouter, **attrs) + self.assertIsNone(result) diff --git a/openstackclient/tests/unit/network/v2/test_security_group.py b/openstackclient/tests/unit/network/v2/test_security_group.py index 43aa07cc..9a30267e 100644 --- a/openstackclient/tests/unit/network/v2/test_security_group.py +++ b/openstackclient/tests/unit/network/v2/test_security_group.py @@ -456,7 +456,7 @@ class TestListSecurityGroupNetwork(TestSecurityGroupNetwork): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) - filters = {'tenant_id': project.id} + filters = {'tenant_id': project.id, 'project_id': project.id} self.network.security_groups.assert_called_once_with(**filters) self.assertEqual(self.columns, columns) @@ -476,7 +476,7 @@ class TestListSecurityGroupNetwork(TestSecurityGroupNetwork): parsed_args = self.check_parser(self.cmd, arglist, verifylist) columns, data = self.cmd.take_action(parsed_args) - filters = {'tenant_id': project.id} + filters = {'tenant_id': project.id, 'project_id': project.id} self.network.security_groups.assert_called_once_with(**filters) self.assertEqual(self.columns, columns) diff --git a/openstackclient/tests/unit/volume/v2/fakes.py b/openstackclient/tests/unit/volume/v2/fakes.py index d5cd72ec..a6676403 100644 --- a/openstackclient/tests/unit/volume/v2/fakes.py +++ b/openstackclient/tests/unit/volume/v2/fakes.py @@ -903,3 +903,23 @@ class FakeType(object): volume_types.append(volume_type) return volume_types + + @staticmethod + def get_types(types=None, count=2): + """Get an iterable MagicMock object with a list of faked types. + + If types list is provided, then initialize the Mock object with the + list. Otherwise create one. + + :param List types: + A list of FakeResource objects faking types + :param Integer count: + The number of types to be faked + :return + An iterable Mock object with side_effect set to a list of faked + types + """ + if types is None: + types = FakeType.create_types(count) + + return mock.Mock(side_effect=types) diff --git a/openstackclient/tests/unit/volume/v2/test_backup.py b/openstackclient/tests/unit/volume/v2/test_backup.py index 10e7aac5..a8e81c7e 100644 --- a/openstackclient/tests/unit/volume/v2/test_backup.py +++ b/openstackclient/tests/unit/volume/v2/test_backup.py @@ -418,6 +418,30 @@ class TestBackupSet(TestBackup): self.backup.id, **{'name': 'new_name'}) self.assertIsNone(result) + def test_backup_set_description(self): + arglist = [ + '--description', 'new_description', + self.backup.id, + ] + verifylist = [ + ('name', None), + ('description', 'new_description'), + ('backup', self.backup.id), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'description': 'new_description' + } + self.backups_mock.update.assert_called_once_with( + self.backup.id, + **kwargs + ) + self.assertIsNone(result) + def test_backup_set_state(self): arglist = [ '--state', 'error', diff --git a/openstackclient/tests/unit/volume/v2/test_consistency_group.py b/openstackclient/tests/unit/volume/v2/test_consistency_group.py index bc99ca8d..6eeeae39 100644 --- a/openstackclient/tests/unit/volume/v2/test_consistency_group.py +++ b/openstackclient/tests/unit/volume/v2/test_consistency_group.py @@ -36,10 +36,115 @@ class TestConsistencyGroup(volume_fakes.TestVolume): self.app.client_manager.volume.cgsnapshots) self.cgsnapshots_mock.reset_mock() + self.volumes_mock = ( + self.app.client_manager.volume.volumes) + self.volumes_mock.reset_mock() + self.types_mock = self.app.client_manager.volume.volume_types self.types_mock.reset_mock() +class TestConsistencyGroupAddVolume(TestConsistencyGroup): + + _consistency_group = ( + volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) + + def setUp(self): + super(TestConsistencyGroupAddVolume, self).setUp() + + self.consistencygroups_mock.get.return_value = ( + self._consistency_group) + # Get the command object to test + self.cmd = \ + consistency_group.AddVolumeToConsistencyGroup(self.app, None) + + def test_add_one_volume_to_consistency_group(self): + volume = volume_fakes.FakeVolume.create_one_volume() + self.volumes_mock.get.return_value = volume + arglist = [ + self._consistency_group.id, + volume.id, + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volume.id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'add_volumes': volume.id, + } + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, + **kwargs + ) + self.assertIsNone(result) + + def test_add_multiple_volumes_to_consistency_group(self): + volumes = volume_fakes.FakeVolume.create_volumes(count=2) + self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes) + arglist = [ + self._consistency_group.id, + volumes[0].id, + volumes[1].id, + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volumes[0].id, volumes[1].id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'add_volumes': volumes[0].id + ',' + volumes[1].id, + } + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, + **kwargs + ) + self.assertIsNone(result) + + @mock.patch.object(consistency_group.LOG, 'error') + def test_add_multiple_volumes_to_consistency_group_with_exception( + self, mock_error): + volume = volume_fakes.FakeVolume.create_one_volume() + arglist = [ + self._consistency_group.id, + volume.id, + 'unexist_volume', + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volume.id, 'unexist_volume']), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [volume, + exceptions.CommandError, + self._consistency_group] + with mock.patch.object(utils, 'find_resource', + side_effect=find_mock_result) as find_mock: + result = self.cmd.take_action(parsed_args) + mock_error.assert_called_with("1 of 2 volumes failed to add.") + self.assertIsNone(result) + find_mock.assert_any_call(self.consistencygroups_mock, + self._consistency_group.id) + find_mock.assert_any_call(self.volumes_mock, + volume.id) + find_mock.assert_any_call(self.volumes_mock, + 'unexist_volume') + self.assertEqual(3, find_mock.call_count) + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, add_volumes=volume.id + ) + + class TestConsistencyGroupCreate(TestConsistencyGroup): volume_type = volume_fakes.FakeType.create_one_type() @@ -394,6 +499,107 @@ class TestConsistencyGroupList(TestConsistencyGroup): self.assertEqual(self.data_long, list(data)) +class TestConsistencyGroupRemoveVolume(TestConsistencyGroup): + + _consistency_group = ( + volume_fakes.FakeConsistencyGroup.create_one_consistency_group()) + + def setUp(self): + super(TestConsistencyGroupRemoveVolume, self).setUp() + + self.consistencygroups_mock.get.return_value = ( + self._consistency_group) + # Get the command object to test + self.cmd = \ + consistency_group.RemoveVolumeFromConsistencyGroup(self.app, None) + + def test_remove_one_volume_from_consistency_group(self): + volume = volume_fakes.FakeVolume.create_one_volume() + self.volumes_mock.get.return_value = volume + arglist = [ + self._consistency_group.id, + volume.id, + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volume.id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'remove_volumes': volume.id, + } + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, + **kwargs + ) + self.assertIsNone(result) + + def test_remove_multi_volumes_from_consistency_group(self): + volumes = volume_fakes.FakeVolume.create_volumes(count=2) + self.volumes_mock.get = volume_fakes.FakeVolume.get_volumes(volumes) + arglist = [ + self._consistency_group.id, + volumes[0].id, + volumes[1].id, + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volumes[0].id, volumes[1].id]), + ] + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + result = self.cmd.take_action(parsed_args) + + # Set expected values + kwargs = { + 'remove_volumes': volumes[0].id + ',' + volumes[1].id, + } + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, + **kwargs + ) + self.assertIsNone(result) + + @mock.patch.object(consistency_group.LOG, 'error') + def test_remove_multiple_volumes_from_consistency_group_with_exception( + self, mock_error): + volume = volume_fakes.FakeVolume.create_one_volume() + arglist = [ + self._consistency_group.id, + volume.id, + 'unexist_volume', + ] + verifylist = [ + ('consistency_group', self._consistency_group.id), + ('volumes', [volume.id, 'unexist_volume']), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [volume, + exceptions.CommandError, + self._consistency_group] + with mock.patch.object(utils, 'find_resource', + side_effect=find_mock_result) as find_mock: + result = self.cmd.take_action(parsed_args) + mock_error.assert_called_with("1 of 2 volumes failed to remove.") + self.assertIsNone(result) + find_mock.assert_any_call(self.consistencygroups_mock, + self._consistency_group.id) + find_mock.assert_any_call(self.volumes_mock, + volume.id) + find_mock.assert_any_call(self.volumes_mock, + 'unexist_volume') + self.assertEqual(3, find_mock.call_count) + self.consistencygroups_mock.update.assert_called_once_with( + self._consistency_group.id, remove_volumes=volume.id + ) + + class TestConsistencyGroupSet(TestConsistencyGroup): consistency_group = ( diff --git a/openstackclient/tests/unit/volume/v2/test_qos_specs.py b/openstackclient/tests/unit/volume/v2/test_qos_specs.py index 7597e852..35d9a345 100644 --- a/openstackclient/tests/unit/volume/v2/test_qos_specs.py +++ b/openstackclient/tests/unit/volume/v2/test_qos_specs.py @@ -70,24 +70,26 @@ class TestQosAssociate(TestQos): class TestQosCreate(TestQos): - new_qos_spec = volume_fakes.FakeQos.create_one_qos() columns = ( 'consumer', 'id', 'name', - 'specs' - ) - data = ( - new_qos_spec.consumer, - new_qos_spec.id, - new_qos_spec.name, - new_qos_spec.specs + 'properties' ) def setUp(self): super(TestQosCreate, self).setUp() + self.new_qos_spec = volume_fakes.FakeQos.create_one_qos() self.qos_mock.create.return_value = self.new_qos_spec + + self.data = ( + self.new_qos_spec.consumer, + self.new_qos_spec.id, + self.new_qos_spec.name, + utils.format_dict(self.new_qos_spec.specs) + ) + # Get the command object to test self.cmd = qos_specs.CreateQos(self.app, None) @@ -147,11 +149,11 @@ class TestQosCreate(TestQos): columns, data = self.cmd.take_action(parsed_args) - self.new_qos_spec.specs.update( - {'consumer': self.new_qos_spec.consumer}) self.qos_mock.create.assert_called_with( self.new_qos_spec.name, - self.new_qos_spec.specs + {'consumer': self.new_qos_spec.consumer, + 'foo': 'bar', + 'iops': '9001'} ) self.assertEqual(self.columns, columns) @@ -307,7 +309,7 @@ class TestQosList(TestQos): 'Name', 'Consumer', 'Associations', - 'Specs', + 'Properties', ) data = [] for q in qos_specs: @@ -383,7 +385,7 @@ class TestQosShow(TestQos): 'consumer', 'id', 'name', - 'specs' + 'properties' ) data = ( qos_association.name, diff --git a/openstackclient/tests/unit/volume/v2/test_snapshot.py b/openstackclient/tests/unit/volume/v2/test_snapshot.py index 8ce356ae..12d1e390 100644 --- a/openstackclient/tests/unit/volume/v2/test_snapshot.py +++ b/openstackclient/tests/unit/volume/v2/test_snapshot.py @@ -19,6 +19,7 @@ from mock import call from osc_lib import exceptions from osc_lib import utils +from openstackclient.tests.unit.identity.v3 import fakes as project_fakes from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes from openstackclient.volume.v2 import volume_snapshot @@ -32,6 +33,8 @@ class TestSnapshot(volume_fakes.TestVolume): self.snapshots_mock.reset_mock() self.volumes_mock = self.app.client_manager.volume.volumes self.volumes_mock.reset_mock() + self.project_mock = self.app.client_manager.identity.projects + self.project_mock.reset_mock() class TestSnapshotCreate(TestSnapshot): @@ -278,6 +281,7 @@ class TestSnapshotDelete(TestSnapshot): class TestSnapshotList(TestSnapshot): volume = volume_fakes.FakeVolume.create_one_volume() + project = project_fakes.FakeProject.create_one_project() snapshots = volume_fakes.FakeSnapshot.create_snapshots( attrs={'volume_id': volume.name}, count=3) @@ -321,6 +325,7 @@ class TestSnapshotList(TestSnapshot): self.volumes_mock.list.return_value = [self.volume] self.volumes_mock.get.return_value = self.volume + self.project_mock.get.return_value = self.project self.snapshots_mock.list.return_value = self.snapshots # Get the command to test self.cmd = volume_snapshot.ListVolumeSnapshot(self.app, None) @@ -341,6 +346,7 @@ class TestSnapshotList(TestSnapshot): 'all_tenants': False, 'name': None, 'status': None, + 'project_id': None, 'volume_id': None } ) @@ -351,11 +357,13 @@ class TestSnapshotList(TestSnapshot): arglist = [ "--long", "--limit", "2", + "--project", self.project.id, "--marker", self.snapshots[0].id, ] verifylist = [ ("long", True), ("limit", 2), + ("project", self.project.id), ("marker", self.snapshots[0].id), ('all_projects', False), ] @@ -367,7 +375,8 @@ class TestSnapshotList(TestSnapshot): limit=2, marker=self.snapshots[0].id, search_opts={ - 'all_tenants': False, + 'all_tenants': True, + 'project_id': self.project.id, 'name': None, 'status': None, 'volume_id': None @@ -394,6 +403,7 @@ class TestSnapshotList(TestSnapshot): 'all_tenants': True, 'name': None, 'status': None, + 'project_id': None, 'volume_id': None } ) @@ -419,6 +429,7 @@ class TestSnapshotList(TestSnapshot): 'all_tenants': False, 'name': self.snapshots[0].name, 'status': None, + 'project_id': None, 'volume_id': None } ) @@ -444,6 +455,7 @@ class TestSnapshotList(TestSnapshot): 'all_tenants': False, 'name': None, 'status': self.snapshots[0].status, + 'project_id': None, 'volume_id': None } ) @@ -469,6 +481,7 @@ class TestSnapshotList(TestSnapshot): 'all_tenants': False, 'name': None, 'status': None, + 'project_id': None, 'volume_id': self.volume.id } ) diff --git a/openstackclient/tests/unit/volume/v2/test_type.py b/openstackclient/tests/unit/volume/v2/test_type.py index 0d556e13..cec01bd8 100644 --- a/openstackclient/tests/unit/volume/v2/test_type.py +++ b/openstackclient/tests/unit/volume/v2/test_type.py @@ -13,6 +13,7 @@ # import mock +from mock import call from osc_lib import exceptions from osc_lib import utils @@ -133,12 +134,13 @@ class TestTypeCreate(TestType): class TestTypeDelete(TestType): - volume_type = volume_fakes.FakeType.create_one_type() + volume_types = volume_fakes.FakeType.create_types(count=2) def setUp(self): super(TestTypeDelete, self).setUp() - self.types_mock.get.return_value = self.volume_type + self.types_mock.get = volume_fakes.FakeType.get_types( + self.volume_types) self.types_mock.delete.return_value = None # Get the command object to mock @@ -146,18 +148,64 @@ class TestTypeDelete(TestType): def test_type_delete(self): arglist = [ - self.volume_type.id + self.volume_types[0].id ] verifylist = [ - ("volume_types", [self.volume_type.id]) + ("volume_types", [self.volume_types[0].id]) ] parsed_args = self.check_parser(self.cmd, arglist, verifylist) result = self.cmd.take_action(parsed_args) - self.types_mock.delete.assert_called_with(self.volume_type) + self.types_mock.delete.assert_called_with(self.volume_types[0]) self.assertIsNone(result) + def test_delete_multiple_types(self): + arglist = [] + for t in self.volume_types: + arglist.append(t.id) + verifylist = [ + ('volume_types', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + result = self.cmd.take_action(parsed_args) + + calls = [] + for t in self.volume_types: + calls.append(call(t)) + self.types_mock.delete.assert_has_calls(calls) + self.assertIsNone(result) + + def test_delete_multiple_types_with_exception(self): + arglist = [ + self.volume_types[0].id, + 'unexist_type', + ] + verifylist = [ + ('volume_types', arglist), + ] + + parsed_args = self.check_parser(self.cmd, arglist, verifylist) + + find_mock_result = [self.volume_types[0], exceptions.CommandError] + with mock.patch.object(utils, 'find_resource', + side_effect=find_mock_result) as find_mock: + try: + self.cmd.take_action(parsed_args) + self.fail('CommandError should be raised.') + except exceptions.CommandError as e: + self.assertEqual('1 of 2 volume types failed to delete.', + str(e)) + find_mock.assert_any_call( + self.types_mock, self.volume_types[0].id) + find_mock.assert_any_call(self.types_mock, 'unexist_type') + + self.assertEqual(2, find_mock.call_count) + self.types_mock.delete.assert_called_once_with( + self.volume_types[0] + ) + class TestTypeList(TestType): diff --git a/openstackclient/tests/unit/volume/v2/test_volume.py b/openstackclient/tests/unit/volume/v2/test_volume.py index 1529c2e2..4fef9dd9 100644 --- a/openstackclient/tests/unit/volume/v2/test_volume.py +++ b/openstackclient/tests/unit/volume/v2/test_volume.py @@ -823,6 +823,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': None, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -853,6 +866,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': True, + 'project_id': self.project.id, + 'user_id': None, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -885,6 +911,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': True, + 'project_id': self.project.id, + 'user_id': None, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -915,6 +954,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': self.user.id, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] device = self.mock_volume.attachments[0]['device'] @@ -946,6 +998,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': self.user.id, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -976,6 +1041,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': None, + 'display_name': self.mock_volume.name, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -1006,6 +1084,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': None, + 'display_name': None, + 'status': self.mock_volume.status, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -1036,6 +1127,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': True, + 'project_id': None, + 'user_id': None, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + self.assertEqual(self.columns, columns) server = self.mock_volume.attachments[0]['server_id'] @@ -1067,6 +1171,19 @@ class TestVolumeList(TestVolume): columns, data = self.cmd.take_action(parsed_args) + search_opts = { + 'all_tenants': False, + 'project_id': None, + 'user_id': None, + 'display_name': None, + 'status': None, + } + self.volumes_mock.list.assert_called_once_with( + search_opts=search_opts, + marker=None, + limit=None, + ) + collist = [ 'ID', 'Display Name', |