Merge "Move volume v2 service clients to v3 dir"

This commit is contained in:
Zuul 2018-05-23 23:41:01 +00:00 committed by Gerrit Code Review
commit 11a62d47e6
41 changed files with 1681 additions and 1307 deletions

View File

@ -0,0 +1,12 @@
---
features:
- |
Adds volume service clients for v3 APIs. As v3 base API should be
identical to v2 APIs, we just copy all existing v2 service client
for v3 API.
deprecations:
- |
Deprecates the volume service clients for v2 APIs. Volume v2 APIs
are deprecated in all supported stable branches, so it's time
to deprecate the tempest service clients for v2 APIs and remove in future
release.

View File

@ -13,15 +13,11 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_serialization import jsonutils as json
from debtcollector import moves
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import availability_zone_client
class AvailabilityZoneClient(rest_client.RestClient):
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)
AvailabilityZoneClient = moves.moved_class(
availability_zone_client.AvailabilityZoneClient, 'AvailabilityZoneClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,107 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume import base_client
from tempest.lib.services.volume.v3 import backups_client
class BackupsClient(base_client.BaseClient):
"""Volume Backups client"""
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://developer.openstack.org/api-ref/block-storage/v2/index.html#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://developer.openstack.org/api-ref/block-storage/v2/index.html#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, **params):
"""List all the tenant's backups.
For a full list of available parameters, please refer to the official
API reference:
http://developer.openstack.org/api-ref/block-storage/v2/#list-backups
http://developer.openstack.org/api-ref/block-storage/v2/#list-backups-with-details
"""
url = "backups"
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 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."""
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'
BackupsClient = moves.moved_class(
backups_client.BackupsClient, 'BackupsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -13,22 +13,11 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_serialization import jsonutils as json
from debtcollector import moves
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import capabilities_client
class CapabilitiesClient(rest_client.RestClient):
def show_backend_capabilities(self, host):
"""Shows capabilities for a storage back end.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/index.html#show-back-end-capabilities
"""
url = 'capabilities/%s' % host
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
CapabilitiesClient = moves.moved_class(
capabilities_client.CapabilitiesClient, 'CapabilitiesClient',
__name__, version="Queens", removal_version='?')

View File

@ -13,78 +13,11 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_serialization import jsonutils as json
from debtcollector import moves
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume.v3 import encryption_types_client
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.
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 show_encryption_specs_item(self, volume_type_id, key):
"""Get the encryption specs item for the specified volume type."""
url = "/types/%s/encryption/%s" % (volume_type_id, key)
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://developer.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)
def update_encryption_type(self, volume_type_id, **kwargs):
"""Update an encryption type for an existing volume type.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/#update-an-encryption-type-for-v2
"""
url = "/types/%s/encryption/provider" % volume_type_id
put_body = json.dumps({'encryption': kwargs})
resp, body = self.put(url, put_body)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
EncryptionTypesClient = moves.moved_class(
encryption_types_client.EncryptionTypesClient, 'EncryptionTypesClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,18 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import extensions_client
class ExtensionsClient(rest_client.RestClient):
"""Volume 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)
ExtensionsClient = moves.moved_class(
extensions_client.ExtensionsClient, 'ExtensionsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,36 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import hosts_client
class HostsClient(rest_client.RestClient):
"""Client class to send CRUD Volume API requests"""
def list_hosts(self, **params):
"""Lists all hosts.
For a full list of available parameters, please refer to the official
API reference:
https://developer.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)
def show_host(self, host_name):
"""Show host details."""
url = 'os-hosts/%s' % host_name
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
HostsClient = moves.moved_class(
hosts_client.HostsClient, 'HostsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,19 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import limits_client
class LimitsClient(rest_client.RestClient):
"""Volume limits client."""
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)
LimitsClient = moves.moved_class(
limits_client.LimitsClient, 'LimitsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -11,121 +11,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume.v3 import qos_client
class QosSpecsClient(rest_client.RestClient):
"""Volume QoS client.
Client class to send CRUD QoS API requests
"""
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:
http://developer.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:
http://developer.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:
http://developer.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)
QosSpecsClient = moves.moved_class(
qos_client.QosSpecsClient, 'QosSpecsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,38 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import quota_classes_client
class QuotaClassesClient(rest_client.RestClient):
"""Volume quota class client."""
def show_quota_class_set(self, quota_class_id):
"""List quotas for a quota class.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/index.html#show-quota-classes
"""
url = 'os-quota-class-sets/%s' % quota_class_id
resp, body = self.get(url)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
def update_quota_class_set(self, quota_class_id, **kwargs):
"""Update quotas for a quota class.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/index.html#update-quota-classes
"""
url = 'os-quota-class-sets/%s' % quota_class_id
put_body = json.dumps({'quota_class_set': kwargs})
resp, body = self.put(url, put_body)
self.expected_success(200, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
QuotaClassesClient = moves.moved_class(
quota_classes_client.QuotaClassesClient, 'QuotaClassesClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,52 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import quotas_client
class QuotasClient(rest_client.RestClient):
"""Client class to send CRUD Volume Quotas API 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://developer.openstack.org/api-ref/block-storage/v2/index.html#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)
QuotasClient = moves.moved_class(
quotas_client.QuotasClient, 'QuotasClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,25 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import scheduler_stats_client
class SchedulerStatsClient(rest_client.RestClient):
def list_pools(self, detail=False):
"""List all the volumes pools (hosts).
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/index.html#list-back-end-storage-pools
"""
url = 'scheduler-stats/get_pools'
if detail:
url += '?detail=True'
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
SchedulerStatsClient = moves.moved_class(
scheduler_stats_client.SchedulerStatsClient, 'SchedulerStatsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,22 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import services_client
class ServicesClient(rest_client.RestClient):
"""Client class to send CRUD Volume API requests"""
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)
ServicesClient = moves.moved_class(
services_client.ServicesClient, 'ServicesClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,20 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import snapshot_manage_client
class SnapshotManageClient(rest_client.RestClient):
"""Snapshot manage client."""
def manage_snapshot(self, **kwargs):
"""Manage a snapshot."""
post_body = json.dumps({'snapshot': kwargs})
url = 'os-snapshot-manage'
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
SnapshotManageClient = moves.moved_class(
snapshot_manage_client.SnapshotManageClient, 'SnapshotManageClient',
__name__, version="Rocky", removal_version='?')

View File

@ -9,199 +9,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume.v3 import snapshots_client
class SnapshotsClient(rest_client.RestClient):
"""Client class to send CRUD Volume API requests."""
create_resp = 202
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:
http://developer.openstack.org/api-ref/block-storage/v2/#list-snapshots-with-details
http://developer.openstack.org/api-ref/block-storage/v2/#list-snapshots
"""
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:
http://developer.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:
http://developer.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 update_snapshot(self, snapshot_id, **kwargs):
"""Updates a snapshot.
For a full list of available parameters, please refer to the official
API reference:
http://developer.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 delete_snapshot(self, snapshot_id):
"""Delete Snapshot.
For a full list of available parameters, please refer to the official
API reference:
http://developer.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.
For a full list of available parameters, please refer to the official
API reference:
http://developer.openstack.org/api-ref/block-storage/v2/#create-snapshot-metadata
"""
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 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:
http://developer.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:
http://developer.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 show_snapshot_metadata_item(self, snapshot_id, id):
"""Show metadata item for the snapshot."""
url = "snapshots/%s/metadata/%s" % (snapshot_id, 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_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)
def unmanage_snapshot(self, snapshot_id):
"""Unmanage a snapshot."""
post_body = json.dumps({'os-unmanage': {}})
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)
SnapshotsClient = moves.moved_class(
snapshots_client.SnapshotsClient, 'SnapshotsClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,71 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import transfers_client
class TransfersClient(rest_client.RestClient):
"""Client class to send CRUD Volume Transfer API requests"""
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://developer.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, detail=False, **params):
"""List all the volume transfers created.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/#list-volume-transfers
https://developer.openstack.org/api-ref/block-storage/v2/#list-volume-transfers-with-details
"""
url = 'os-volume-transfer'
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 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://developer.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)
TransfersClient = moves.moved_class(
transfers_client.TransfersClient, 'TransfersClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,193 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume.v3 import types_client
class TypesClient(rest_client.RestClient):
"""Client class to send CRUD Volume 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://developer.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://developer.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://developer.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:
http://developer.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:
http://developer.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.
volume_type_id: Id of volume_type.
extra_spec_name: Name of the extra spec to be updated.
extra_spec: 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:
http://developer.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)
def add_type_access(self, volume_type_id, **kwargs):
"""Adds volume type access for the given project.
For a full list of available parameters, please refer to the official
API reference:
http://developer.openstack.org/api-ref/block-storage/v2/#add-private-volume-type-access
"""
post_body = json.dumps({'addProjectAccess': kwargs})
url = 'types/%s/action' % volume_type_id
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def remove_type_access(self, volume_type_id, **kwargs):
"""Removes volume type access for the given project.
For a full list of available parameters, please refer to the official
API reference:
http://developer.openstack.org/api-ref/block-storage/v2/#remove-private-volume-type-access
"""
post_body = json.dumps({'removeProjectAccess': kwargs})
url = 'types/%s/action' % volume_type_id
resp, body = self.post(url, post_body)
self.expected_success(202, resp.status)
return rest_client.ResponseBody(resp, body)
def list_type_access(self, volume_type_id):
"""Print access information about the given volume type.
For a full list of available parameters, please refer to the official
API reference:
http://developer.openstack.org/api-ref/block-storage/v2/#list-private-volume-type-access-details
"""
url = 'types/%s/os-volume-type-access' % volume_type_id
resp, body = self.get(url)
body = json.loads(body)
self.expected_success(200, resp.status)
return rest_client.ResponseBody(resp, body)
TypesClient = moves.moved_class(
types_client.TypesClient, 'TypesClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,24 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
from tempest.lib.common import rest_client
from tempest.lib.services.volume.v3 import volume_manage_client
class VolumeManageClient(rest_client.RestClient):
"""Volume manage client."""
def manage_volume(self, **kwargs):
"""Manage existing volume.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/#manage-existing-volume
"""
post_body = json.dumps({'volume': kwargs})
resp, body = self.post('os-volume-manage', post_body)
self.expected_success(202, resp.status)
body = json.loads(body)
return rest_client.ResponseBody(resp, body)
VolumeManageClient = moves.moved_class(
volume_manage_client.VolumeManageClient, 'VolumeManageClient',
__name__, version="Rocky", removal_version='?')

View File

@ -12,340 +12,11 @@
# 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 debtcollector import moves
from oslo_serialization import jsonutils as json
import six
from six.moves.urllib import parse as urllib
from tempest.lib.common import rest_client
from tempest.lib import exceptions as lib_exc
from tempest.lib.services.volume import base_client
from tempest.lib.services.volume.v3 import volumes_client
class VolumesClient(base_client.BaseClient):
"""Client class to send CRUD Volume 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, six.string_types):
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:
http://developer.openstack.org/api-ref/block-storage/v2/#list-volumes-with-details
http://developer.openstack.org/api-ref/block-storage/v2/#list-volumes
"""
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:
http://developer.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(202, 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:
http://developer.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, **params):
"""Deletes the Specified Volume.
For a full list of available parameters, please refer to the official
API reference:
https://developer.openstack.org/api-ref/block-storage/v2/#delete-volume
"""
url = 'volumes/%s' % volume_id
if params:
url += '?%s' % urllib.urlencode(params)
resp, body = self.delete(url)
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)