Remove volume v1 service clients

Cinder has removed the volume v1 APIs in queens
cycle: I03bf2db5bd7e2fdfb4f6032758ccaf2b348a82ba

Tempest does not support queens release so we are good
to remove the volume v1 service clients also.

Change-Id: I297f230de51e0ef4f35eb33ddbaaab53c230713f
This commit is contained in:
Ghanshyam Mann 2021-05-18 14:31:05 -05:00
parent 393e94a604
commit 08d5376e5b
22 changed files with 10 additions and 1637 deletions

View File

@ -0,0 +1,8 @@
---
prelude: >
Tempest Service clients for volume v1 APIs are removed.
upgrade:
- |
Cinder removed the volume v1 APIs in queens release and Tempest
now support only stable train onwards release so all the Tempest
service clients for volume v1 APIs are now removed.

View File

@ -52,7 +52,6 @@ def tempest_modules():
'image.v2': image.v2,
'network': network,
'object-storage': object_storage,
'volume.v1': volume.v1,
'volume.v2': volume.v2,
'volume.v3': volume.v3
}

View File

@ -12,8 +12,7 @@
# License for the specific language governing permissions and limitations under
# the License.
from tempest.lib.services.volume import v1
from tempest.lib.services.volume import v2
from tempest.lib.services.volume import v3
__all__ = ['v1', 'v2', 'v3']
__all__ = ['v2', 'v3']

View File

@ -1,33 +0,0 @@
# Copyright (c) 2016 Hewlett-Packard Enterprise Development Company, L.P.
#
# 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 tempest.lib.services.volume.v1.availability_zone_client \
import AvailabilityZoneClient
from tempest.lib.services.volume.v1.backups_client import BackupsClient
from tempest.lib.services.volume.v1.encryption_types_client import \
EncryptionTypesClient
from tempest.lib.services.volume.v1.extensions_client import ExtensionsClient
from tempest.lib.services.volume.v1.hosts_client import HostsClient
from tempest.lib.services.volume.v1.limits_client import LimitsClient
from tempest.lib.services.volume.v1.qos_client import QosSpecsClient
from tempest.lib.services.volume.v1.quotas_client import QuotasClient
from tempest.lib.services.volume.v1.services_client import ServicesClient
from tempest.lib.services.volume.v1.snapshots_client import SnapshotsClient
from tempest.lib.services.volume.v1.types_client import TypesClient
from tempest.lib.services.volume.v1.volumes_client import VolumesClient
__all__ = ['AvailabilityZoneClient', 'BackupsClient', 'EncryptionTypesClient',
'ExtensionsClient', 'HostsClient', 'QosSpecsClient', 'QuotasClient',
'ServicesClient', 'SnapshotsClient', 'TypesClient', 'VolumesClient',
'LimitsClient']

View File

@ -1,28 +0,0 @@
# Copyright 2014 NEC 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class AvailabilityZoneClient(rest_client.RestClient):
"""Volume V1 availability zone client."""
def list_availability_zones(self):
resp, body = self.get('os-availability-zone')
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,112 +0,0 @@
# Copyright 2014 OpenStack Foundation
# 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class BackupsClient(rest_client.RestClient):
"""Volume V1 Backups client"""
api_version = "v1"
def create_backup(self, **kwargs):
"""Creates a backup of volume.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-backup
"""
post_body = json.dumps({'backup': kwargs})
resp, body = self.post('backups', post_body)
body = json.loads(body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def restore_backup(self, backup_id, **kwargs):
"""Restore volume from backup.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#restore-backup
"""
post_body = json.dumps({'restore': kwargs})
resp, body = self.post('backups/%s/restore' % (backup_id), post_body)
body = json.loads(body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_backup(self, backup_id):
"""Delete a backup of volume."""
resp, body = self.delete('backups/%s' % backup_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def show_backup(self, backup_id):
"""Returns the details of a single backup."""
url = "backups/%s" % backup_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def list_backups(self, detail=False):
"""Information for all the tenant's backups."""
url = "backups"
if detail:
url += "/detail"
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def export_backup(self, backup_id):
"""Export backup metadata record."""
url = "backups/%s/export_record" % backup_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def import_backup(self, **kwargs):
"""Import backup metadata record."""
# TODO(linanbj): Current api-site doesn't contain this API description.
# After fixing the api-site, we need to fix here also for putting the
# link to api-site.
post_body = json.dumps({'backup-record': kwargs})
resp, body = self.post("backups/import_record", post_body)
body = json.loads(body)
self.expected_success(201, resp.status)
return rest_client.ResponseBody(resp, body)
def reset_backup_status(self, backup_id, status):
"""Reset the specified backup's status."""
post_body = json.dumps({'os-reset_status': {"status": status}})
resp, body = self.post('backups/%s/action' % backup_id, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def is_resource_deleted(self, id):
try:
self.show_backup(id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'backup'

View File

@ -1,68 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class EncryptionTypesClient(rest_client.RestClient):
def is_resource_deleted(self, id):
try:
body = self.show_encryption_type(id)
if not body:
return True
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'encryption-type'
def show_encryption_type(self, volume_type_id):
"""Get the volume encryption type for the specified volume type.
:param volume_type_id: Id of volume type.
"""
url = "/types/%s/encryption" % volume_type_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def create_encryption_type(self, volume_type_id, **kwargs):
"""Create encryption type.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-an-encryption-type-for-v2
"""
url = "/types/%s/encryption" % volume_type_id
post_body = json.dumps({'encryption': kwargs})
resp, body = self.post(url, post_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_encryption_type(self, volume_type_id):
"""Delete the encryption type for the specified volume type."""
resp, body = self.delete(
"/types/%s/encryption/provider" % volume_type_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,29 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class ExtensionsClient(rest_client.RestClient):
"""Volume V1 extensions client."""
def list_extensions(self):
url = 'extensions'
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,40 +0,0 @@
# Copyright 2013 OpenStack Foundation
# 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 urllib import parse as urllib
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class HostsClient(rest_client.RestClient):
"""Client class to send CRUD Volume Host API V1 requests"""
def list_hosts(self, **params):
"""Lists all hosts.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#list-all-hosts
"""
url = 'os-hosts'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,32 +0,0 @@
# Copyright 2016 Red Hat, Inc.
# 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class LimitsClient(rest_client.RestClient):
"""Volume V1 limits client."""
api_version = "v1"
def show_limits(self):
"""Returns the details of a volume absolute limits."""
url = "limits"
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,133 +0,0 @@
# 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 oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class QosSpecsClient(rest_client.RestClient):
"""Volume V1 QoS client.
Client class to send CRUD QoS API requests
"""
api_version = "v1"
def is_resource_deleted(self, qos_id):
try:
self.show_qos(qos_id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'qos'
def create_qos(self, **kwargs):
"""Create a QoS Specification.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-qos-specification
"""
post_body = json.dumps({'qos_specs': kwargs})
resp, body = self.post('qos-specs', post_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_qos(self, qos_id, force=False):
"""Delete the specified QoS specification."""
resp, body = self.delete(
"qos-specs/%s?force=%s" % (qos_id, force))
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def list_qos(self):
"""List all the QoS specifications created."""
url = 'qos-specs'
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_qos(self, qos_id):
"""Get the specified QoS specification."""
url = "qos-specs/%s" % qos_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def set_qos_key(self, qos_id, **kwargs):
"""Set the specified keys/values of QoS specification.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#set-keys-in-qos-specification
"""
put_body = json.dumps({"qos_specs": kwargs})
resp, body = self.put('qos-specs/%s' % qos_id, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def unset_qos_key(self, qos_id, keys):
"""Unset the specified keys of QoS specification.
:param keys: keys to delete from the QoS specification.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#unset-keys-in-qos-specification
"""
put_body = json.dumps({'keys': keys})
resp, body = self.put('qos-specs/%s/delete_keys' % qos_id, put_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def associate_qos(self, qos_id, vol_type_id):
"""Associate the specified QoS with specified volume-type."""
url = "qos-specs/%s/associate" % qos_id
url += "?vol_type_id=%s" % vol_type_id
resp, body = self.get(url)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def show_association_qos(self, qos_id):
"""Get the association of the specified QoS specification."""
url = "qos-specs/%s/associations" % qos_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def disassociate_qos(self, qos_id, vol_type_id):
"""Disassociate the specified QoS with specified volume-type."""
url = "qos-specs/%s/disassociate" % qos_id
url += "?vol_type_id=%s" % vol_type_id
resp, body = self.get(url)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def disassociate_all_qos(self, qos_id):
"""Disassociate the specified QoS with all associations."""
url = "qos-specs/%s/disassociate_all" % qos_id
resp, body = self.get(url)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,63 +0,0 @@
# Copyright (C) 2014 eNovance SAS <licensing@enovance.com>
#
# 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 urllib import parse as urllib
from oslo_serialization import jsonutils
from tempest.lib.common import rest_client
class QuotasClient(rest_client.RestClient):
"""Client class to send CRUD Volume Quotas API V1 requests"""
def show_default_quota_set(self, tenant_id):
"""List the default volume quota set for a tenant."""
url = 'os-quota-sets/%s/defaults' % tenant_id
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def show_quota_set(self, tenant_id, params=None):
"""List the quota set for a tenant."""
url = 'os-quota-sets/%s' % tenant_id
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def update_quota_set(self, tenant_id, **kwargs):
"""Updates quota set
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-quotas
"""
put_body = jsonutils.dumps({'quota_set': kwargs})
resp, body = self.put('os-quota-sets/%s' % tenant_id, put_body)
self.expected_success(200, resp.status)
body = jsonutils.loads(body)
return rest_client.ResponseBody(resp, body)
def delete_quota_set(self, tenant_id):
"""Delete the tenant's quota set."""
resp, body = self.delete('os-quota-sets/%s' % tenant_id)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,34 +0,0 @@
# Copyright 2014 NEC 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 urllib import parse as urllib
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
class ServicesClient(rest_client.RestClient):
"""Volume V1 volume services client"""
def list_services(self, **params):
url = 'os-services'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,188 +0,0 @@
# 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 urllib import parse as urllib
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class SnapshotsClient(rest_client.RestClient):
"""Client class to send CRUD Volume V1 API requests."""
create_resp = 200
def list_snapshots(self, detail=False, **params):
"""List all the snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#list-snapshots
https://docs.openstack.org/api-ref/block-storage/v2/#list-snapshots-with-details
"""
url = 'snapshots'
if detail:
url += '/detail'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_snapshot(self, snapshot_id):
"""Returns the details of a single snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#show-snapshot-details
"""
url = "snapshots/%s" % snapshot_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def create_snapshot(self, **kwargs):
"""Creates a new snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-snapshot
"""
post_body = json.dumps({'snapshot': kwargs})
resp, body = self.post('snapshots', post_body)
body = json.loads(body)
self.expected_success(self.create_resp, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_snapshot(self, snapshot_id):
"""Delete Snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#delete-snapshot
"""
resp, body = self.delete("snapshots/%s" % snapshot_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def is_resource_deleted(self, id):
try:
self.show_snapshot(id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'volume-snapshot'
def reset_snapshot_status(self, snapshot_id, status):
"""Reset the specified snapshot's status."""
post_body = json.dumps({'os-reset_status': {"status": status}})
resp, body = self.post('snapshots/%s/action' % snapshot_id, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def update_snapshot_status(self, snapshot_id, **kwargs):
"""Update the specified snapshot's status."""
# TODO(gmann): api-site doesn't contain doc ref
# for this API. After fixing the api-site, we need to
# add the link here.
# Bug https://bugs.launchpad.net/openstack-api-site/+bug/1532645
post_body = json.dumps({'os-update_snapshot_status': kwargs})
url = 'snapshots/%s/action' % snapshot_id
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def create_snapshot_metadata(self, snapshot_id, metadata):
"""Create metadata for the snapshot."""
put_body = json.dumps({'metadata': metadata})
url = "snapshots/%s/metadata" % snapshot_id
resp, body = self.post(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_snapshot(self, snapshot_id, **kwargs):
"""Updates a snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-snapshot
"""
put_body = json.dumps({'snapshot': kwargs})
resp, body = self.put('snapshots/%s' % snapshot_id, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_snapshot_metadata(self, snapshot_id):
"""Get metadata of the snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#show-snapshot-metadata
"""
url = "snapshots/%s/metadata" % snapshot_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_snapshot_metadata(self, snapshot_id, **kwargs):
"""Update metadata for the snapshot.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-snapshot-metadata
"""
put_body = json.dumps(kwargs)
url = "snapshots/%s/metadata" % snapshot_id
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_snapshot_metadata_item(self, snapshot_id, id, **kwargs):
"""Update metadata item for the snapshot."""
# TODO(piyush): Current api-site doesn't contain this API description.
# After fixing the api-site, we need to fix here also for putting the
# link to api-site.
# LP: https://bugs.launchpad.net/openstack-api-site/+bug/1529064
put_body = json.dumps(kwargs)
url = "snapshots/%s/metadata/%s" % (snapshot_id, id)
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_snapshot_metadata_item(self, snapshot_id, id):
"""Delete metadata item for the snapshot."""
url = "snapshots/%s/metadata/%s" % (snapshot_id, id)
resp, body = self.delete(url)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def force_delete_snapshot(self, snapshot_id):
"""Force Delete Snapshot."""
post_body = json.dumps({'os-force_delete': {}})
resp, body = self.post('snapshots/%s/action' % snapshot_id, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,167 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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 urllib import parse as urllib
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class TypesClient(rest_client.RestClient):
"""Client class to send CRUD Volume Types API requests"""
def is_resource_deleted(self, id):
try:
self.show_volume_type(id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'volume-type'
def list_volume_types(self, **params):
"""List all the volume_types created.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#list-all-volume-types-for-v2
"""
url = 'types'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_volume_type(self, volume_type_id):
"""Returns the details of a single volume_type.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#show-volume-type-details-for-v2
"""
url = "types/%s" % volume_type_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def create_volume_type(self, **kwargs):
"""Create volume type.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-volume-type-for-v2
"""
post_body = json.dumps({'volume_type': kwargs})
resp, body = self.post('types', post_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_volume_type(self, volume_type_id):
"""Deletes the Specified Volume_type.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#delete-volume-type
"""
resp, body = self.delete("types/%s" % volume_type_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def list_volume_types_extra_specs(self, volume_type_id, **params):
"""List all the volume_types extra specs created.
TODO: Current api-site doesn't contain this API description.
After fixing the api-site, we need to fix here also for putting
the link to api-site.
"""
url = 'types/%s/extra_specs' % volume_type_id
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_volume_type_extra_specs(self, volume_type_id, extra_specs_name):
"""Returns the details of a single volume_type extra spec."""
url = "types/%s/extra_specs/%s" % (volume_type_id, extra_specs_name)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def create_volume_type_extra_specs(self, volume_type_id, extra_specs):
"""Creates a new Volume_type extra spec.
volume_type_id: Id of volume_type.
extra_specs: A dictionary of values to be used as extra_specs.
"""
url = "types/%s/extra_specs" % volume_type_id
post_body = json.dumps({'extra_specs': extra_specs})
resp, body = self.post(url, post_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_volume_type_extra_specs(self, volume_type_id, extra_spec_name):
"""Deletes the Specified Volume_type extra spec."""
resp, body = self.delete("types/%s/extra_specs/%s" % (
volume_type_id, extra_spec_name))
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume_type(self, volume_type_id, **kwargs):
"""Updates volume type name, description, and/or is_public.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-volume-type
"""
put_body = json.dumps({'volume_type': kwargs})
resp, body = self.put('types/%s' % volume_type_id, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume_type_extra_specs(self, volume_type_id, extra_spec_name,
extra_specs):
"""Update a volume_type extra spec.
:param volume_type_id: Id of volume_type.
:param extra_spec_name: Name of the extra spec to be updated.
:param extra_specs: A dictionary of with key as extra_spec_name and the
updated value.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-extra-specs-for-a-volume-type
"""
url = "types/%s/extra_specs/%s" % (volume_type_id, extra_spec_name)
put_body = json.dumps(extra_specs)
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)

View File

@ -1,306 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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 urllib import parse as urllib
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
class VolumesClient(rest_client.RestClient):
"""Client class to send CRUD Volume V1 API requests"""
def _prepare_params(self, params):
"""Prepares params for use in get or _ext_get methods.
If params is a string it will be left as it is, but if it's not it will
be urlencoded.
"""
if isinstance(params, str):
return params
return urllib.urlencode(params)
def list_volumes(self, detail=False, params=None):
"""List all the volumes created.
Params can be a string (must be urlencoded) or a dictionary.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#list-volumes
https://docs.openstack.org/api-ref/block-storage/v2/#list-volumes-with-details
"""
url = 'volumes'
if detail:
url += '/detail'
if params:
url += '?%s' % self._prepare_params(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_volume(self, volume_id):
"""Returns the details of a single volume."""
url = "volumes/%s" % volume_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def create_volume(self, **kwargs):
"""Creates a new Volume.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-volume
"""
post_body = json.dumps({'volume': kwargs})
resp, body = self.post('volumes', post_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume(self, volume_id, **kwargs):
"""Updates the Specified Volume.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#update-volume
"""
put_body = json.dumps({'volume': kwargs})
resp, body = self.put('volumes/%s' % volume_id, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_volume(self, volume_id):
"""Deletes the Specified Volume."""
resp, body = self.delete("volumes/%s" % volume_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def upload_volume(self, volume_id, **kwargs):
"""Uploads a volume in Glance."""
post_body = json.dumps({'os-volume_upload_image': kwargs})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
body = json.loads(body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def attach_volume(self, volume_id, **kwargs):
"""Attaches a volume to a given instance on a given mountpoint.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#attach-volume-to-server
"""
post_body = json.dumps({'os-attach': kwargs})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def set_bootable_volume(self, volume_id, **kwargs):
"""set a bootable flag for a volume - true or false."""
post_body = json.dumps({'os-set_bootable': kwargs})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def detach_volume(self, volume_id):
"""Detaches a volume from an instance."""
post_body = json.dumps({'os-detach': {}})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def reserve_volume(self, volume_id):
"""Reserves a volume."""
post_body = json.dumps({'os-reserve': {}})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def unreserve_volume(self, volume_id):
"""Restore a reserved volume ."""
post_body = json.dumps({'os-unreserve': {}})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def is_resource_deleted(self, id):
try:
self.show_volume(id)
except lib_exc.NotFound:
return True
return False
@property
def resource_type(self):
"""Returns the primary type of resource this client works with."""
return 'volume'
def extend_volume(self, volume_id, **kwargs):
"""Extend a volume.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#extend-volume-size
"""
post_body = json.dumps({'os-extend': kwargs})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def reset_volume_status(self, volume_id, **kwargs):
"""Reset the Specified Volume's Status.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#reset-volume-statuses
"""
post_body = json.dumps({'os-reset_status': kwargs})
resp, body = self.post('volumes/%s/action' % volume_id, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def create_volume_transfer(self, **kwargs):
"""Create a volume transfer.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#create-volume-transfer
"""
post_body = json.dumps({'transfer': kwargs})
resp, body = self.post('os-volume-transfer', post_body)
body = json.loads(body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def show_volume_transfer(self, transfer_id):
"""Returns the details of a volume transfer."""
url = "os-volume-transfer/%s" % transfer_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def list_volume_transfers(self, **params):
"""List all the volume transfers created.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#list-volume-transfers
"""
url = 'os-volume-transfer'
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_volume_transfer(self, transfer_id):
"""Delete a volume transfer."""
resp, body = self.delete("os-volume-transfer/%s" % transfer_id)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def accept_volume_transfer(self, transfer_id, **kwargs):
"""Accept a volume transfer.
For a full list of available parameters, please refer to the official
API reference:
https://docs.openstack.org/api-ref/block-storage/v2/#accept-volume-transfer
"""
url = 'os-volume-transfer/%s/accept' % transfer_id
post_body = json.dumps({'accept': kwargs})
resp, body = self.post(url, post_body)
body = json.loads(body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume_readonly(self, volume_id, **kwargs):
"""Update the Specified Volume readonly."""
post_body = json.dumps({'os-update_readonly_flag': kwargs})
url = 'volumes/%s/action' % (volume_id)
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def force_delete_volume(self, volume_id):
"""Force Delete Volume."""
post_body = json.dumps({'os-force_delete': {}})
resp, body = self.post('volumes/%s/action' % volume_id, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def create_volume_metadata(self, volume_id, metadata):
"""Create metadata for the volume."""
put_body = json.dumps({'metadata': metadata})
url = "volumes/%s/metadata" % volume_id
resp, body = self.post(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def show_volume_metadata(self, volume_id):
"""Get metadata of the volume."""
url = "volumes/%s/metadata" % volume_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume_metadata(self, volume_id, metadata):
"""Update metadata for the volume."""
put_body = json.dumps({'metadata': metadata})
url = "volumes/%s/metadata" % volume_id
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def update_volume_metadata_item(self, volume_id, id, meta_item):
"""Update metadata item for the volume."""
put_body = json.dumps({'meta': meta_item})
url = "volumes/%s/metadata/%s" % (volume_id, id)
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def delete_volume_metadata_item(self, volume_id, id):
"""Delete metadata item for the volume."""
url = "volumes/%s/metadata/%s" % (volume_id, id)
resp, body = self.delete(url)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
def retype_volume(self, volume_id, **kwargs):
"""Updates volume with new volume type."""
post_body = json.dumps({'os-retype': kwargs})
resp, _ = self.post('volumes/%s/action' % volume_id, post_body)
self.expected_success(202, resp.status)

View File

@ -38,8 +38,7 @@ class RegistryFixture(fixtures.Fixture):
"""Initialise the registry fixture"""
self.services = set(['compute', 'identity.v2', 'identity.v3',
'image.v1', 'image.v2', 'network', 'placement',
'volume.v1', 'volume.v2', 'volume.v3',
'object-storage'])
'volume.v2', 'volume.v3', 'object-storage'])
def _setUp(self):
# Cleanup the registry

View File

@ -1,86 +0,0 @@
# Copyright 2016 NEC 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 tempest.lib.services.volume.v1 import encryption_types_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestEncryptionTypesClient(base.BaseServiceTest):
FAKE_CREATE_ENCRYPTION_TYPE = {
"encryption": {
"id": "cbc36478b0bd8e67e89",
"name": "FakeEncryptionType",
"type": "fakeType",
"provider": "LuksEncryptor",
"cipher": "aes-xts-plain64",
"key_size": "512",
"control_location": "front-end"
}
}
FAKE_INFO_ENCRYPTION_TYPE = {
"encryption": {
"name": "FakeEncryptionType",
"type": "fakeType",
"description": "test_description",
"volume_type": "fakeType",
"provider": "LuksEncryptor",
"cipher": "aes-xts-plain64",
"key_size": "512",
"control_location": "front-end"
}
}
def setUp(self):
super(TestEncryptionTypesClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = encryption_types_client.EncryptionTypesClient(fake_auth,
'volume',
'regionOne'
)
def _test_create_encryption(self, bytes_body=False):
self.check_service_client_function(
self.client.create_encryption_type,
'tempest.lib.common.rest_client.RestClient.post',
self.FAKE_CREATE_ENCRYPTION_TYPE,
bytes_body, volume_type_id="cbc36478b0bd8e67e89")
def _test_show_encryption_type(self, bytes_body=False):
self.check_service_client_function(
self.client.show_encryption_type,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_INFO_ENCRYPTION_TYPE,
bytes_body, volume_type_id="cbc36478b0bd8e67e89")
def test_create_encryption_type_with_str_body(self):
self._test_create_encryption()
def test_create_encryption_type_with_bytes_body(self):
self._test_create_encryption(bytes_body=True)
def test_show_encryption_type_with_str_body(self):
self._test_show_encryption_type()
def test_show_encryption_type_with_bytes_body(self):
self._test_show_encryption_type(bytes_body=True)
def test_delete_encryption_type(self):
self.check_service_client_function(
self.client.delete_encryption_type,
'tempest.lib.common.rest_client.RestClient.delete',
{},
volume_type_id="cbc36478b0bd8e67e89",
status=202)

View File

@ -1,96 +0,0 @@
# Copyright 2016 NEC 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 tempest.lib.services.volume.v1 import quotas_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestQuotasClient(base.BaseServiceTest):
FAKE_QUOTAS = {
"quota_set": {
"cores": 20,
"fixed_ips": -1,
"floating_ips": 10,
"id": "fake_tenant",
"injected_file_content_bytes": 10240,
"injected_file_path_bytes": 255,
"injected_files": 5,
"instances": 10,
"key_pairs": 100,
"metadata_items": 128,
"ram": 51200,
"security_group_rules": 20,
"security_groups": 10
}
}
FAKE_UPDATE_QUOTAS_REQUEST = {
"quota_set": {
"security_groups": 45
}
}
def setUp(self):
super(TestQuotasClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = quotas_client.QuotasClient(fake_auth,
'volume',
'regionOne')
def _test_show_default_quota_set(self, bytes_body=False):
self.check_service_client_function(
self.client.show_default_quota_set,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_QUOTAS,
bytes_body, tenant_id="fake_tenant")
def _test_show_quota_set(self, bytes_body=False):
self.check_service_client_function(
self.client.show_quota_set,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_QUOTAS,
bytes_body, tenant_id="fake_tenant")
def _test_update_quota_set(self, bytes_body=False):
self.check_service_client_function(
self.client.update_quota_set,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_UPDATE_QUOTAS_REQUEST,
bytes_body, tenant_id="fake_tenant")
def test_show_default_quota_set_with_str_body(self):
self._test_show_default_quota_set()
def test_show_default_quota_set_with_bytes_body(self):
self._test_show_default_quota_set(bytes_body=True)
def test_show_quota_set_with_str_body(self):
self._test_show_quota_set()
def test_show_quota_set_with_bytes_body(self):
self._test_show_quota_set(bytes_body=True)
def test_update_quota_set_with_str_body(self):
self._test_update_quota_set()
def test_update_quota_set_with_bytes_body(self):
self._test_update_quota_set(bytes_body=True)
def test_delete_quota_set(self):
self.check_service_client_function(
self.client.delete_quota_set,
'tempest.lib.common.rest_client.RestClient.delete',
{},
tenant_id="fake_tenant")

View File

@ -1,200 +0,0 @@
# Copyright 2016 NEC 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 tempest.lib.services.volume.v1 import snapshots_client
from tempest.tests.lib import fake_auth_provider
from tempest.tests.lib.services import base
class TestSnapshotsClient(base.BaseServiceTest):
FAKE_CREATE_SNAPSHOT = {
"snapshot": {
"display_name": "snap-001",
"display_description": "Daily backup",
"volume_id": "521752a6-acf6-4b2d-bc7a-119f9148cd8c",
"force": True
}
}
FAKE_UPDATE_SNAPSHOT_REQUEST = {
"metadata": {
"key": "v1"
}
}
FAKE_INFO_SNAPSHOT = {
"snapshot": {
"id": "3fbbcccf-d058-4502-8844-6feeffdf4cb5",
"display_name": "snap-001",
"display_description": "Daily backup",
"volume_id": "521752a6-acf6-4b2d-bc7a-119f9148cd8c",
"status": "available",
"size": 30,
"created_at": "2012-02-29T03:50:07Z"
}
}
FAKE_LIST_SNAPSHOTS = {
"snapshots": [
{
"id": "3fbbcccf-d058-4502-8844-6feeffdf4cb5",
"display_name": "snap-001",
"display_description": "Daily backup",
"volume_id": "521752a6-acf6-4b2d-bc7a-119f9148cd8c",
"status": "available",
"size": 30,
"created_at": "2012-02-29T03:50:07Z",
"metadata": {
"contents": "junk"
}
},
{
"id": "e479997c-650b-40a4-9dfe-77655818b0d2",
"display_name": "snap-002",
"display_description": "Weekly backup",
"volume_id": "76b8950a-8594-4e5b-8dce-0dfa9c696358",
"status": "available",
"size": 25,
"created_at": "2012-03-19T01:52:47Z",
"metadata": {}
}
]
}
def setUp(self):
super(TestSnapshotsClient, self).setUp()
fake_auth = fake_auth_provider.FakeAuthProvider()
self.client = snapshots_client.SnapshotsClient(fake_auth,
'volume',
'regionOne')
def _test_create_snapshot(self, bytes_body=False):
self.check_service_client_function(
self.client.create_snapshot,
'tempest.lib.common.rest_client.RestClient.post',
self.FAKE_CREATE_SNAPSHOT,
bytes_body)
def _test_show_snapshot(self, bytes_body=False):
self.check_service_client_function(
self.client.show_snapshot,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_INFO_SNAPSHOT,
bytes_body,
snapshot_id="3fbbcccf-d058-4502-8844-6feeffdf4cb5")
def _test_list_snapshots(self, bytes_body=False):
self.check_service_client_function(
self.client.list_snapshots,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_LIST_SNAPSHOTS,
bytes_body,
detail=True)
def _test_create_snapshot_metadata(self, bytes_body=False):
self.check_service_client_function(
self.client.create_snapshot_metadata,
'tempest.lib.common.rest_client.RestClient.post',
self.FAKE_INFO_SNAPSHOT,
bytes_body,
snapshot_id="3fbbcccf-d058-4502-8844-6feeffdf4cb5",
metadata={"key": "v1"})
def _test_update_snapshot(self, bytes_body=False):
self.check_service_client_function(
self.client.update_snapshot,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_UPDATE_SNAPSHOT_REQUEST,
bytes_body,
snapshot_id="3fbbcccf-d058-4502-8844-6feeffdf4cb5")
def _test_show_snapshot_metadata(self, bytes_body=False):
self.check_service_client_function(
self.client.show_snapshot_metadata,
'tempest.lib.common.rest_client.RestClient.get',
self.FAKE_UPDATE_SNAPSHOT_REQUEST,
bytes_body,
snapshot_id="3fbbcccf-d058-4502-8844-6feeffdf4cb5")
def _test_update_snapshot_metadata(self, bytes_body=False):
self.check_service_client_function(
self.client.update_snapshot_metadata,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_UPDATE_SNAPSHOT_REQUEST,
bytes_body, snapshot_id="cbc36478b0bd8e67e89")
def _test_update_snapshot_metadata_item(self, bytes_body=False):
self.check_service_client_function(
self.client.update_snapshot_metadata_item,
'tempest.lib.common.rest_client.RestClient.put',
self.FAKE_INFO_SNAPSHOT,
bytes_body, volume_type_id="cbc36478b0bd8e67e89")
def test_create_snapshot_with_str_body(self):
self._test_create_snapshot()
def test_create_snapshot_with_bytes_body(self):
self._test_create_snapshot(bytes_body=True)
def test_show_snapshot_with_str_body(self):
self._test_show_snapshot()
def test_show_snapshot_with_bytes_body(self):
self._test_show_snapshot(bytes_body=True)
def test_list_snapshots_with_str_body(self):
self._test_list_snapshots()
def test_list_snapshots_with_bytes_body(self):
self._test_list_snapshots(bytes_body=True)
def test_create_snapshot_metadata_with_str_body(self):
self._test_create_snapshot_metadata()
def test_create_snapshot_metadata_with_bytes_body(self):
self._test_create_snapshot_metadata(bytes_body=True)
def test_update_snapshot_with_str_body(self):
self._test_update_snapshot()
def test_update_snapshot_with_bytes_body(self):
self._test_update_snapshot(bytes_body=True)
def test_show_snapshot_metadata_with_str_body(self):
self._test_show_snapshot_metadata()
def test_show_snapshot_metadata_with_bytes_body(self):
self._test_show_snapshot_metadata(bytes_body=True)
def test_update_snapshot_metadata_with_str_body(self):
self._test_update_snapshot_metadata()
def test_update_snapshot_metadata_with_bytes_body(self):
self._test_update_snapshot_metadata(bytes_body=True)
def test_force_delete_snapshot(self):
self.check_service_client_function(
self.client.force_delete_snapshot,
'tempest.lib.common.rest_client.RestClient.post',
{},
snapshot_id="521752a6-acf6-4b2d-bc7a-119f9148cd8c",
status=202)
def test_delete_snapshot(self):
self.check_service_client_function(
self.client.delete_snapshot,
'tempest.lib.common.rest_client.RestClient.delete',
{},
snapshot_id="521752a6-acf6-4b2d-bc7a-119f9148cd8c",
status=202)

View File

@ -21,21 +21,6 @@ class TestVersionsClient(base.BaseServiceTest):
FAKE_VERSIONS_INFO = {
"versions": [
{
"status": "DEPRECATED", "updated": "2016-05-02T20:25:19Z",
"links": [
{"href": "http://docs.openstack.org/", "type": "text/html",
"rel": "describedby"},
{"href": "https://10.30.197.39:8776/v1/", "rel": "self"}
],
"min_version": "",
"version": "",
"media-types": [
{"base": "application/json",
"type": "application/vnd.openstack.volume+json;version=1"}
],
"id": "v1.0"
},
{
"status": "DEPRECATED", "updated": "2017-02-25T12:00:00Z",
"links": [
@ -134,8 +119,6 @@ class TestVersionsClient(base.BaseServiceTest):
self._test_show_version('v3', bytes_body=True)
def test_get_base_version_url_app_name(self):
self._test_get_base_version_url('https://bar.org/volume/v1/123',
'https://bar.org/volume/')
self._test_get_base_version_url('https://bar.org/volume/v2/123',
'https://bar.org/volume/')
self._test_get_base_version_url('https://bar.org/volume/v3/123',