diff options
Diffstat (limited to 'openstackclient/tests/functional')
13 files changed, 1127 insertions, 62 deletions
diff --git a/openstackclient/tests/functional/compute/v2/test_agent.py b/openstackclient/tests/functional/compute/v2/test_agent.py index 1a112e82..25d8c868 100644 --- a/openstackclient/tests/functional/compute/v2/test_agent.py +++ b/openstackclient/tests/functional/compute/v2/test_agent.py @@ -21,10 +21,10 @@ class ComputeAgentTests(base.TestCase): # Generate two different md5hash MD5HASH1 = hashlib.md5() - MD5HASH1.update('agent_1') + MD5HASH1.update('agent_1'.encode('utf-8')) MD5HASH1 = MD5HASH1.hexdigest() MD5HASH2 = hashlib.md5() - MD5HASH2.update('agent_2') + MD5HASH2.update('agent_2'.encode('utf-8')) MD5HASH2 = MD5HASH2.hexdigest() def test_compute_agent_delete(self): diff --git a/openstackclient/tests/functional/compute/v2/test_aggregate.py b/openstackclient/tests/functional/compute/v2/test_aggregate.py index 8a082e82..1de53099 100644 --- a/openstackclient/tests/functional/compute/v2/test_aggregate.py +++ b/openstackclient/tests/functional/compute/v2/test_aggregate.py @@ -11,7 +11,6 @@ # under the License. import json -import time import uuid from openstackclient.tests.functional import base @@ -20,35 +19,14 @@ from openstackclient.tests.functional import base class AggregateTests(base.TestCase): """Functional tests for aggregate""" - @classmethod - def wait_for_status(cls, check_type, check_name, desired_status, - wait=120, interval=5, failures=None): - current_status = "notset" - if failures is None: - failures = ['error'] - total_sleep = 0 - while total_sleep < wait: - output = json.loads(cls.openstack( - check_type + ' show -f json ' + check_name)) - current_status = output['name'] - if (current_status == desired_status): - print('{} {} now has status {}' - .format(check_type, check_name, current_status)) - return - print('Checking {} {} Waiting for {} current status: {}' - .format(check_type, check_name, - desired_status, current_status)) - if current_status in failures: - raise Exception( - 'Current status {} of {} {} is one of failures {}' - .format(current_status, check_type, check_name, failures)) - time.sleep(interval) - total_sleep += interval - cls.assertOutput(desired_status, current_status) - def test_aggregate_crud(self): """Test create, delete multiple""" name1 = uuid.uuid4().hex + self.addCleanup( + self.openstack, + 'aggregate delete ' + name1, + fail_ok=True, + ) cmd_output = json.loads(self.openstack( 'aggregate create -f json ' + '--zone nova ' + @@ -67,14 +45,16 @@ class AggregateTests(base.TestCase): 'a', cmd_output['properties'] ) - self.wait_for_status('aggregate', name1, name1) + cmd_output = json.loads(self.openstack( + 'aggregate show -f json ' + name1)) + self.assertEqual(name1, cmd_output['name']) + + name2 = uuid.uuid4().hex self.addCleanup( self.openstack, - 'aggregate delete ' + name1, + 'aggregate delete ' + name2, fail_ok=True, ) - - name2 = uuid.uuid4().hex cmd_output = json.loads(self.openstack( 'aggregate create -f json ' + '--zone external ' + @@ -88,15 +68,17 @@ class AggregateTests(base.TestCase): 'external', cmd_output['availability_zone'] ) - self.wait_for_status('aggregate', name2, name2) + cmd_output = json.loads(self.openstack( + 'aggregate show -f json ' + name2)) + self.assertEqual(name2, cmd_output['name']) + + # Test aggregate set + name3 = uuid.uuid4().hex self.addCleanup( self.openstack, - 'aggregate delete ' + name2, + 'aggregate delete ' + name3, fail_ok=True, ) - - # Test aggregate set - name3 = uuid.uuid4().hex raw_output = self.openstack( 'aggregate set ' + '--name ' + name3 + ' ' + @@ -106,12 +88,6 @@ class AggregateTests(base.TestCase): name1 ) self.assertOutput('', raw_output) - self.wait_for_status('aggregate', name3, name3) - self.addCleanup( - self.openstack, - 'aggregate delete ' + name3, - fail_ok=True, - ) cmd_output = json.loads(self.openstack( 'aggregate show -f json ' + @@ -198,11 +174,11 @@ class AggregateTests(base.TestCase): self.skipTest("Skip aggregates in a Nova cells v1 configuration") name = uuid.uuid4().hex + self.addCleanup(self.openstack, 'aggregate delete ' + name) self.openstack( 'aggregate create ' + name ) - self.addCleanup(self.openstack, 'aggregate delete ' + name) # Test add host cmd_output = json.loads(self.openstack( diff --git a/openstackclient/tests/functional/compute/v2/test_keypair.py b/openstackclient/tests/functional/compute/v2/test_keypair.py index 9a88e66f..42f334a4 100644 --- a/openstackclient/tests/functional/compute/v2/test_keypair.py +++ b/openstackclient/tests/functional/compute/v2/test_keypair.py @@ -88,7 +88,7 @@ class KeypairTests(KeypairBase): 1) Create keypair with given public key 2) Delete keypair """ - with tempfile.NamedTemporaryFile() as f: + with tempfile.NamedTemporaryFile(mode='w+') as f: f.write(self.PUBLIC_KEY) f.flush() @@ -108,7 +108,7 @@ class KeypairTests(KeypairBase): 1) Create keypair with private key file 2) Delete keypair """ - with tempfile.NamedTemporaryFile() as f: + with tempfile.NamedTemporaryFile(mode='w+') as f: cmd_output = json.loads(self.openstack( 'keypair create -f json --private-key %s tmpkey' % f.name, )) diff --git a/openstackclient/tests/functional/compute/v2/test_server.py b/openstackclient/tests/functional/compute/v2/test_server.py index 2bca70d0..6e080e9b 100644 --- a/openstackclient/tests/functional/compute/v2/test_server.py +++ b/openstackclient/tests/functional/compute/v2/test_server.py @@ -63,6 +63,49 @@ class ServerTests(common.ComputeTestCase): self.assertNotIn(name1, col_name) self.assertIn(name2, col_name) + def test_server_list_with_marker_and_deleted(self): + """Test server list with deleted and marker""" + cmd_output = self.server_create(cleanup=False) + name1 = cmd_output['name'] + cmd_output = self.server_create(cleanup=False) + name2 = cmd_output['name'] + id2 = cmd_output['id'] + self.wait_for_status(name1, "ACTIVE") + self.wait_for_status(name2, "ACTIVE") + + # Test list --marker with ID + cmd_output = json.loads(self.openstack( + 'server list -f json --marker ' + id2 + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + + # Test list --marker with Name + cmd_output = json.loads(self.openstack( + 'server list -f json --marker ' + name2 + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + + self.openstack('server delete --wait ' + name1) + self.openstack('server delete --wait ' + name2) + + # Test list --deleted --marker with ID + cmd_output = json.loads(self.openstack( + 'server list -f json --deleted --marker ' + id2 + )) + col_name = [x["Name"] for x in cmd_output] + self.assertIn(name1, col_name) + + # Test list --deleted --marker with Name + try: + cmd_output = json.loads(self.openstack( + 'server list -f json --deleted --marker ' + name2 + )) + except exceptions.CommandFailed as e: + self.assertIn('marker [%s] not found (HTTP 400)' % (name2), + e.stderr.decode('utf-8')) + def test_server_list_with_changes_before(self): """Test server list. diff --git a/openstackclient/tests/functional/identity/v2/common.py b/openstackclient/tests/functional/identity/v2/common.py index f4bc10bd..43c0cbf2 100644 --- a/openstackclient/tests/functional/identity/v2/common.py +++ b/openstackclient/tests/functional/identity/v2/common.py @@ -11,6 +11,7 @@ # under the License. import os +import unittest import fixtures from tempest.lib.common.utils import data_utils @@ -62,7 +63,7 @@ class IdentityTests(base.TestCase): # TODO(dtroyer): Actually determine if Identity v2 admin is # enabled in the target cloud. Tuens out OSC # doesn't make this easy as it should (yet). - raise cls.skipException('No Identity v2 admin endpoint?') + raise unittest.case.SkipTest('No Identity v2 admin endpoint?') @classmethod def tearDownClass(cls): diff --git a/openstackclient/tests/functional/identity/v3/common.py b/openstackclient/tests/functional/identity/v3/common.py index 43b416aa..a5edd9a5 100644 --- a/openstackclient/tests/functional/identity/v3/common.py +++ b/openstackclient/tests/functional/identity/v3/common.py @@ -33,7 +33,7 @@ class IdentityTests(base.TestCase): 'password_expires_at'] PROJECT_FIELDS = ['description', 'id', 'domain_id', 'is_domain', 'enabled', 'name', 'parent_id'] - ROLE_FIELDS = ['id', 'name', 'domain_id'] + ROLE_FIELDS = ['id', 'name', 'domain_id', 'description'] SERVICE_FIELDS = ['id', 'enabled', 'name', 'type', 'description'] REGION_FIELDS = ['description', 'enabled', 'parent_region', 'region'] ENDPOINT_FIELDS = ['id', 'region', 'region_id', 'service_id', @@ -360,7 +360,7 @@ class IdentityTests(base.TestCase): def _extract_value_from_items(self, key, items): for d in items: - for k, v in d.iteritems(): + for k, v in d.items(): if k == key: return v diff --git a/openstackclient/tests/functional/identity/v3/test_role.py b/openstackclient/tests/functional/identity/v3/test_role.py index 38bfff71..3954c4e3 100644 --- a/openstackclient/tests/functional/identity/v3/test_role.py +++ b/openstackclient/tests/functional/identity/v3/test_role.py @@ -20,6 +20,21 @@ class RoleTests(common.IdentityTests): def test_role_create(self): self._create_dummy_role() + def test_role_create_with_description(self): + role_name = data_utils.rand_name('TestRole') + description = data_utils.rand_name('description') + raw_output = self.openstack( + 'role create ' + '--description %(description)s ' + '%(name)s' % {'description': description, + 'name': role_name}) + role = self.parse_show_as_object(raw_output) + self.addCleanup(self.openstack, 'role delete %s' % role['id']) + items = self.parse_show(raw_output) + self.assert_show_fields(items, self.ROLE_FIELDS) + self.assertEqual(description, role['description']) + return role_name + def test_role_delete(self): role_name = self._create_dummy_role(add_clean_up=False) raw_output = self.openstack('role delete %s' % role_name) @@ -47,6 +62,16 @@ class RoleTests(common.IdentityTests): role = self.parse_show_as_object(raw_output) self.assertEqual(new_role_name, role['name']) + def test_role_set_description(self): + role_name = self._create_dummy_role() + description = data_utils.rand_name("NewDescription") + raw_output = self.openstack('role set --description %s %s' + % (description, role_name)) + self.assertEqual(0, len(raw_output)) + raw_output = self.openstack('role show %s' % role_name) + role = self.parse_show_as_object(raw_output) + self.assertEqual(description, role['description']) + def test_role_add(self): role_name = self._create_dummy_role() username = self._create_dummy_user() diff --git a/openstackclient/tests/functional/object/v1/test_object.py b/openstackclient/tests/functional/object/v1/test_object.py index 226ef8ad..b3f23e52 100644 --- a/openstackclient/tests/functional/object/v1/test_object.py +++ b/openstackclient/tests/functional/object/v1/test_object.py @@ -33,7 +33,7 @@ class ObjectTests(common.ObjectStoreTests): self.skipTest("No object-store service present") def test_object(self): - with tempfile.NamedTemporaryFile() as f: + with tempfile.NamedTemporaryFile(mode='w+') as f: f.write('test content') f.flush() self._test_object(f.name) diff --git a/openstackclient/tests/functional/volume/v3/test_qos.py b/openstackclient/tests/functional/volume/v3/test_qos.py index a6290fc5..fdfa6827 100644 --- a/openstackclient/tests/functional/volume/v3/test_qos.py +++ b/openstackclient/tests/functional/volume/v3/test_qos.py @@ -10,9 +10,209 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.volume.v2 import test_qos as v2 +import json +import uuid + from openstackclient.tests.functional.volume.v3 import common -class QosTests(common.BaseVolumeTests, v2.QosTests): +class QosTests(common.BaseVolumeTests): """Functional tests for volume qos. """ + + def test_volume_qos_create_delete_list(self): + """Test create, list, delete multiple""" + name1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume qos create -f json ' + + name1 + )) + self.assertEqual( + name1, + cmd_output['name'] + ) + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume qos create -f json ' + + name2 + )) + self.assertEqual( + name2, + cmd_output['name'] + ) + + # Test list + cmd_output = json.loads(self.openstack( + 'volume qos list -f json' + )) + names = [x["Name"] for x in cmd_output] + self.assertIn(name1, names) + self.assertIn(name2, names) + + # Test delete multiple + del_output = self.openstack('volume qos delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + + def test_volume_qos_set_show_unset(self): + """Tests create volume qos, set, unset, show, delete""" + + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume qos create -f json ' + + '--consumer front-end ' + '--property Alpha=a ' + + name + )) + self.addCleanup(self.openstack, 'volume qos delete ' + name) + self.assertEqual( + name, + cmd_output['name'] + ) + + self.assertEqual( + "front-end", + cmd_output['consumer'] + ) + self.assertEqual( + {'Alpha': 'a'}, + cmd_output['properties'] + ) + + # Test volume qos set + raw_output = self.openstack( + 'volume qos set ' + + '--property Alpha=c ' + + '--property Beta=b ' + + name, + ) + self.assertOutput('', raw_output) + + # Test volume qos show + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + self.assertEqual( + name, + cmd_output['name'] + ) + self.assertEqual( + {'Alpha': 'c', 'Beta': 'b'}, + cmd_output['properties'] + ) + + # Test volume qos unset + raw_output = self.openstack( + 'volume qos unset ' + + '--property Alpha ' + + name, + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + self.assertEqual( + name, + cmd_output['name'] + ) + self.assertEqual( + {'Beta': 'b'}, + cmd_output['properties'] + ) + + def test_volume_qos_asso_disasso(self): + """Tests associate and disassociate qos with volume type""" + vol_type1 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json ' + + vol_type1 + )) + self.assertEqual( + vol_type1, + cmd_output['name'] + ) + self.addCleanup(self.openstack, 'volume type delete ' + vol_type1) + + vol_type2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json ' + + vol_type2 + )) + self.assertEqual( + vol_type2, + cmd_output['name'] + ) + self.addCleanup(self.openstack, 'volume type delete ' + vol_type2) + + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume qos create -f json ' + + name + )) + self.assertEqual( + name, + cmd_output['name'] + ) + self.addCleanup(self.openstack, 'volume qos delete ' + name) + + # Test associate + raw_output = self.openstack( + 'volume qos associate ' + + name + ' ' + vol_type1 + ) + self.assertOutput('', raw_output) + raw_output = self.openstack( + 'volume qos associate ' + + name + ' ' + vol_type2 + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + types = cmd_output["associations"] + self.assertIn(vol_type1, types) + self.assertIn(vol_type2, types) + + # Test disassociate + raw_output = self.openstack( + 'volume qos disassociate ' + + '--volume-type ' + vol_type1 + + ' ' + name + ) + self.assertOutput('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + types = cmd_output["associations"] + self.assertNotIn(vol_type1, types) + self.assertIn(vol_type2, types) + + # Test disassociate --all + raw_output = self.openstack( + 'volume qos associate ' + + name + ' ' + vol_type1 + ) + self.assertOutput('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + types = cmd_output["associations"] + self.assertIn(vol_type1, types) + self.assertIn(vol_type2, types) + + raw_output = self.openstack( + 'volume qos disassociate ' + + '--all ' + name + ) + self.assertOutput('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume qos show -f json ' + + name + )) + self.assertNotIn("associations", cmd_output.keys()) diff --git a/openstackclient/tests/functional/volume/v3/test_transfer_request.py b/openstackclient/tests/functional/volume/v3/test_transfer_request.py index f16dfafa..1bbfedc9 100644 --- a/openstackclient/tests/functional/volume/v3/test_transfer_request.py +++ b/openstackclient/tests/functional/volume/v3/test_transfer_request.py @@ -10,12 +10,112 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.volume.v2 import test_transfer_request \ - as v2 +import json +import uuid + from openstackclient.tests.functional.volume.v3 import common -class TransferRequestTests(common.BaseVolumeTests, v2.TransferRequestTests): +class TransferRequestTests(common.BaseVolumeTests): """Functional tests for transfer request. """ API_VERSION = '3' + + def test_volume_transfer_request_accept(self): + volume_name = uuid.uuid4().hex + xfer_name = uuid.uuid4().hex + + # create a volume + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + volume_name + )) + self.assertEqual(volume_name, cmd_output['name']) + self.addCleanup( + self.openstack, + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume delete ' + + volume_name + ) + self.wait_for_status("volume", volume_name, "available") + + # create volume transfer request for the volume + # and get the auth_key of the new transfer request + cmd_output = json.loads(self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request create -f json ' + + ' --name ' + xfer_name + ' ' + + volume_name + )) + self.assertEqual(xfer_name, cmd_output['name']) + xfer_id = cmd_output['id'] + auth_key = cmd_output['auth_key'] + self.assertTrue(auth_key) + self.wait_for_status("volume", volume_name, "awaiting-transfer") + + # accept the volume transfer request + cmd_output = json.loads(self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request accept -f json ' + + '--auth-key ' + auth_key + ' ' + + xfer_id + )) + self.assertEqual(xfer_name, cmd_output['name']) + self.wait_for_status("volume", volume_name, "available") + + def test_volume_transfer_request_list_show(self): + volume_name = uuid.uuid4().hex + xfer_name = uuid.uuid4().hex + + # create a volume + cmd_output = json.loads(self.openstack( + 'volume create -f json ' + + '--size 1 ' + + volume_name + )) + self.assertEqual(volume_name, cmd_output['name']) + self.addCleanup( + self.openstack, + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume delete ' + + volume_name + ) + self.wait_for_status("volume", volume_name, "available") + + cmd_output = json.loads(self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request create -f json ' + + ' --name ' + xfer_name + ' ' + + volume_name + )) + self.assertEqual(xfer_name, cmd_output['name']) + xfer_id = cmd_output['id'] + auth_key = cmd_output['auth_key'] + self.assertTrue(auth_key) + self.wait_for_status("volume", volume_name, "awaiting-transfer") + + cmd_output = json.loads(self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request list -f json' + )) + self.assertIn(xfer_name, [req['Name'] for req in cmd_output]) + + cmd_output = json.loads(self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request show -f json ' + + xfer_id + )) + self.assertEqual(xfer_name, cmd_output['name']) + + # NOTE(dtroyer): We need to delete the transfer request to allow the + # volume to be deleted. The addCleanup() route does + # not have a mechanism to wait for the volume status + # to become 'available' before attempting to delete + # the volume. + cmd_output = self.openstack( + '--os-volume-api-version ' + self.API_VERSION + ' ' + + 'volume transfer request delete ' + + xfer_id + ) + self.wait_for_status("volume", volume_name, "available") diff --git a/openstackclient/tests/functional/volume/v3/test_volume.py b/openstackclient/tests/functional/volume/v3/test_volume.py index 283b830f..6635167d 100644 --- a/openstackclient/tests/functional/volume/v3/test_volume.py +++ b/openstackclient/tests/functional/volume/v3/test_volume.py @@ -10,9 +10,273 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.volume.v2 import test_volume as v2 +import json +import uuid + from openstackclient.tests.functional.volume.v3 import common -class VolumeTests(common.BaseVolumeTests, v2.VolumeTests): +class VolumeTests(common.BaseVolumeTests): """Functional tests for volume. """ + + 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_status("volume", name1, "available") + self.wait_for_status("volume", name2, "available") + del_output = self.openstack('volume delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + + def test_volume_list(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_status("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_status("volume", name2, "available") + raw_output = self.openstack( + 'volume set ' + + '--state error ' + + name2 + ) + self.assertOutput('', raw_output) + + # Test list --long + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '--long' + )) + names = [x["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["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_and_unset(self): + """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_status("volume", name, "available") + + # Test volume set + raw_output = self.openstack( + 'volume set ' + + '--name ' + new_name + + ' --size 2 ' + + '--description bbbb ' + + '--no-property ' + + '--property Beta=b ' + + '--property Gamma=c ' + + '--image-property a=b ' + + '--image-property c=d ' + + '--bootable ' + + name, + ) + self.assertOutput('', raw_output) + + 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( + {'Beta': 'b', 'Gamma': 'c'}, + cmd_output["properties"], + ) + self.assertEqual( + {'a': 'b', 'c': 'd'}, + cmd_output["volume_image_metadata"], + ) + self.assertEqual( + 'true', + cmd_output["bootable"], + ) + + # Test volume unset + raw_output = self.openstack( + 'volume unset ' + + '--property Beta ' + + '--image-property a ' + + new_name, + ) + self.assertOutput('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume show -f json ' + + new_name + )) + self.assertEqual( + {'Gamma': 'c'}, + cmd_output["properties"], + ) + self.assertEqual( + {'c': 'd'}, + cmd_output["volume_image_metadata"], + ) + + 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_status("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_status("volume snapshot", snapshot_name, "available") + + name = uuid.uuid4().hex + # Create volume from snapshot + 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_status("volume", name, "available") + + # Delete snapshot + raw_output = self.openstack( + 'volume snapshot delete ' + snapshot_name) + self.assertOutput('', raw_output) + # Deleting snapshot may take time. If volume snapshot still exists when + # a parent volume delete is requested, the volume deletion will fail. + self.wait_for_delete('volume snapshot', snapshot_name) + + def test_volume_list_backward_compatibility(self): + """Test backward compatibility of list command""" + 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_status("volume", name1, "available") + + # Test list -c "Display Name" + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '-c "Display Name"' + )) + for each_volume in cmd_output: + self.assertIn('Display Name', each_volume) + + # Test list -c "Name" + cmd_output = json.loads(self.openstack( + 'volume list -f json ' + + '-c "Name"' + )) + for each_volume in cmd_output: + self.assertIn('Name', each_volume) diff --git a/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py b/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py index 28eee6d2..edfdafb6 100644 --- a/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py +++ b/openstackclient/tests/functional/volume/v3/test_volume_snapshot.py @@ -10,9 +10,245 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.volume.v2 import test_volume_snapshot as v2 # noqa +import json +import uuid + from openstackclient.tests.functional.volume.v3 import common -class VolumeSnapshotTests(common.BaseVolumeTests, v2.VolumeSnapshotTests): +class VolumeSnapshotTests(common.BaseVolumeTests): """Functional tests for volume snapshot. """ + + VOLLY = uuid.uuid4().hex + + @classmethod + def setUpClass(cls): + super(VolumeSnapshotTests, cls).setUpClass() + # 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', cls.VOLLY, 'available') + cls.VOLUME_ID = cmd_output['id'] + + @classmethod + def tearDownClass(cls): + try: + cls.wait_for_status('volume', cls.VOLLY, 'available') + raw_output = cls.openstack( + 'volume delete --force ' + cls.VOLLY) + cls.assertOutput('', raw_output) + finally: + super(VolumeSnapshotTests, cls).tearDownClass() + + 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"], + ) + + 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', name1, 'available') + self.wait_for_status('volume snapshot', name2, 'available') + + del_output = self.openstack( + 'volume snapshot delete ' + name1 + ' ' + name2) + self.assertOutput('', del_output) + self.wait_for_delete('volume snapshot', name1) + self.wait_for_delete('volume snapshot', name2) + + 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.wait_for_delete, 'volume snapshot', name1) + 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', name1, 'available') + + name2 = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume snapshot create -f json ' + + name2 + + ' --volume ' + self.VOLLY + )) + self.addCleanup(self.wait_for_delete, 'volume snapshot', name2) + 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', name2, 'available') + raw_output = self.openstack( + '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_volume_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.wait_for_delete, 'volume snapshot', new_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', name, 'available') + + # Test volume snapshot set + raw_output = self.openstack( + '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 snapshot 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"], + ) + + # Test volume snapshot set --no-property + raw_output = self.openstack( + 'volume snapshot set ' + + '--no-property ' + + new_name, + ) + self.assertOutput('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume snapshot show -f json ' + + new_name + )) + self.assertNotIn( + {'Beta': 'b'}, + cmd_output["properties"], + ) diff --git a/openstackclient/tests/functional/volume/v3/test_volume_type.py b/openstackclient/tests/functional/volume/v3/test_volume_type.py index eb66515e..79d40969 100644 --- a/openstackclient/tests/functional/volume/v3/test_volume_type.py +++ b/openstackclient/tests/functional/volume/v3/test_volume_type.py @@ -10,9 +10,229 @@ # License for the specific language governing permissions and limitations # under the License. -from openstackclient.tests.functional.volume.v2 import test_volume_type as v2 +import json +import time +import uuid + from openstackclient.tests.functional.volume.v3 import common -class VolumeTypeTests(common.BaseVolumeTests, v2.VolumeTypeTests): +class VolumeTypeTests(common.BaseVolumeTests): """Functional tests for volume type. """ + + def test_volume_type_create_list(self): + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json --private ' + + name, + )) + self.addCleanup( + self.openstack, + 'volume type delete ' + name, + ) + self.assertEqual(name, cmd_output['name']) + + cmd_output = json.loads(self.openstack( + 'volume type show -f json %s' % name + )) + self.assertEqual(name, cmd_output['name']) + + cmd_output = json.loads(self.openstack('volume type list -f json')) + self.assertIn(name, [t['Name'] for t in cmd_output]) + + cmd_output = json.loads(self.openstack( + 'volume type list -f json --default' + )) + self.assertEqual(1, len(cmd_output)) + self.assertEqual('lvmdriver-1', cmd_output[0]['Name']) + + def test_volume_type_set_unset_properties(self): + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json --private ' + + name, + )) + self.addCleanup( + self.openstack, + 'volume type delete ' + name + ) + self.assertEqual(name, cmd_output['name']) + + raw_output = self.openstack( + 'volume type set --property a=b --property c=d %s' % name + ) + self.assertEqual("", raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json %s' % name + )) + self.assertEqual({'a': 'b', 'c': 'd'}, cmd_output['properties']) + + raw_output = self.openstack( + 'volume type unset --property a %s' % name + ) + self.assertEqual("", raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json %s' % name + )) + self.assertEqual({'c': 'd'}, cmd_output['properties']) + + def test_volume_type_set_unset_multiple_properties(self): + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json --private ' + + name, + )) + self.addCleanup( + self.openstack, + 'volume type delete ' + name + ) + self.assertEqual(name, cmd_output['name']) + + raw_output = self.openstack( + 'volume type set --property a=b --property c=d %s' % name + ) + self.assertEqual("", raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json %s' % name + )) + self.assertEqual({'a': 'b', 'c': 'd'}, cmd_output['properties']) + + raw_output = self.openstack( + 'volume type unset --property a --property c %s' % name + ) + self.assertEqual("", raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json %s' % name + )) + self.assertEqual({}, cmd_output['properties']) + + def test_volume_type_set_unset_project(self): + name = uuid.uuid4().hex + cmd_output = json.loads(self.openstack( + 'volume type create -f json --private ' + + name, + )) + self.addCleanup( + self.openstack, + 'volume type delete ' + name + ) + self.assertEqual(name, cmd_output['name']) + + raw_output = self.openstack( + 'volume type set --project admin %s' % name + ) + self.assertEqual("", raw_output) + + raw_output = self.openstack( + 'volume type unset --project admin %s' % name + ) + self.assertEqual("", raw_output) + + def test_multi_delete(self): + vol_type1 = uuid.uuid4().hex + vol_type2 = uuid.uuid4().hex + self.openstack('volume type create %s' % vol_type1) + time.sleep(5) + self.openstack('volume type create %s' % vol_type2) + time.sleep(5) + cmd = 'volume type delete %s %s' % (vol_type1, vol_type2) + raw_output = self.openstack(cmd) + self.assertOutput('', raw_output) + + # NOTE: Add some basic funtional tests with the old format to + # make sure the command works properly, need to change + # these to new test format when beef up all tests for + # volume tye commands. + def test_encryption_type(self): + name = uuid.uuid4().hex + encryption_type = uuid.uuid4().hex + # test create new encryption type + cmd_output = json.loads(self.openstack( + 'volume type create -f json ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + encryption_type)) + expected = {'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end'} + for attr, value in expected.items(): + self.assertEqual(value, cmd_output['encryption'][attr]) + # test show encryption type + cmd_output = json.loads(self.openstack( + 'volume type show -f json --encryption-type ' + encryption_type)) + expected = {'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end'} + for attr, value in expected.items(): + self.assertEqual(value, cmd_output['encryption'][attr]) + # test list encryption type + cmd_output = json.loads(self.openstack( + 'volume type list -f json --encryption-type')) + encryption_output = [t['Encryption'] for t in cmd_output + if t['Name'] == encryption_type][0] + expected = {'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end'} + for attr, value in expected.items(): + self.assertEqual(value, encryption_output[attr]) + # test set existing encryption type + raw_output = self.openstack( + 'volume type set ' + '--encryption-key-size 256 ' + '--encryption-control-location back-end ' + + encryption_type) + self.assertEqual('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json --encryption-type ' + encryption_type)) + expected = {'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 256, + 'control_location': 'back-end'} + for attr, value in expected.items(): + self.assertEqual(value, cmd_output['encryption'][attr]) + # test set new encryption type + cmd_output = json.loads(self.openstack( + 'volume type create -f json --private ' + + name, + )) + self.addCleanup( + self.openstack, + 'volume type delete ' + name, + ) + self.assertEqual(name, cmd_output['name']) + + raw_output = self.openstack( + 'volume type set ' + '--encryption-provider LuksEncryptor ' + '--encryption-cipher aes-xts-plain64 ' + '--encryption-key-size 128 ' + '--encryption-control-location front-end ' + + name) + self.assertEqual('', raw_output) + + cmd_output = json.loads(self.openstack( + 'volume type show -f json --encryption-type ' + name + )) + expected = {'provider': 'LuksEncryptor', + 'cipher': 'aes-xts-plain64', + 'key_size': 128, + 'control_location': 'front-end'} + for attr, value in expected.items(): + self.assertEqual(value, cmd_output['encryption'][attr]) + # test unset encryption type + raw_output = self.openstack( + 'volume type unset --encryption-type ' + name + ) + self.assertEqual('', raw_output) + cmd_output = json.loads(self.openstack( + 'volume type show -f json --encryption-type ' + name + )) + self.assertEqual({}, cmd_output['encryption']) + # test delete encryption type + raw_output = self.openstack('volume type delete ' + encryption_type) + self.assertEqual('', raw_output) |