openstack/openstacksdk
Code Issues Proposed changes

Blackify openstack.shared_file_system

Browse Source 
Black used with the '-l 79 -S' flags.

A future change will ignore this commit in git-blame history by adding a
'git-blame-ignore-revs' file.

Change-Id: I54209e6cbfeec18a15771882d38730aca273238c
Signed-off-by: Stephen Finucane <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane 2023-05-05 10:59:36 +01:00
parent 82c2a53402
commit 3d2511f980
36 changed files with 768 additions and 600 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
openstack/shared_file_system/shared_file_system_service.py
View File

@ -16,6 +16,7 @@ from openstack.shared_file_system.v2 import _proxy
class SharedFilesystemService(service_description.ServiceDescription):
"""The shared file systems service."""
supported_versions = {
'2': _proxy.Proxy,
}

171
openstack/shared_file_system/v2/_proxy.py
View File

@ -13,34 +13,27 @@
from openstack import proxy
from openstack import resource
from openstack.shared_file_system.v2 import (
availability_zone as _availability_zone)
from openstack.shared_file_system.v2 import (
share_access_rule as _share_access_rule
)
from openstack.shared_file_system.v2 import (
share_export_locations as _share_export_locations
)
from openstack.shared_file_system.v2 import (
share_network as _share_network
)
from openstack.shared_file_system.v2 import (
share_network_subnet as _share_network_subnet
)
from openstack.shared_file_system.v2 import (
share_snapshot as _share_snapshot
)
from openstack.shared_file_system.v2 import (
share_snapshot_instance as _share_snapshot_instance
)
from openstack.shared_file_system.v2 import (
storage_pool as _storage_pool
)
from openstack.shared_file_system.v2 import (
user_message as _user_message
availability_zone as _availability_zone,
)
from openstack.shared_file_system.v2 import limit as _limit
from openstack.shared_file_system.v2 import share as _share
from openstack.shared_file_system.v2 import (
share_access_rule as _share_access_rule,
)
from openstack.shared_file_system.v2 import (
share_export_locations as _share_export_locations,
)
from openstack.shared_file_system.v2 import share_instance as _share_instance
from openstack.shared_file_system.v2 import share_network as _share_network
from openstack.shared_file_system.v2 import (
share_network_subnet as _share_network_subnet,
)
from openstack.shared_file_system.v2 import share_snapshot as _share_snapshot
from openstack.shared_file_system.v2 import (
share_snapshot_instance as _share_snapshot_instance,
)
from openstack.shared_file_system.v2 import storage_pool as _storage_pool
from openstack.shared_file_system.v2 import user_message as _user_message
class Proxy(proxy.Proxy):
@ -53,11 +46,9 @@ class Proxy(proxy.Proxy):
"share": _share.Share,
"share_network": _share_network.ShareNetwork,
"share_network_subnet": _share_network_subnet.ShareNetworkSubnet,
"share_snapshot_instance":
_share_snapshot_instance.ShareSnapshotInstance,
"share_snapshot_instance": _share_snapshot_instance.ShareSnapshotInstance, # noqa: E501
"share_instance": _share_instance.ShareInstance,
"share_export_locations":
_share_export_locations.ShareExportLocation,
"share_export_locations": _share_export_locations.ShareExportLocation,
"share_access_rule": _share_access_rule.ShareAccessRule,
}
@ -134,8 +125,7 @@ class Proxy(proxy.Proxy):
:returns: Result of the ``delete``
:rtype: ``None``
"""
self._delete(_share.Share, share,
ignore_missing=ignore_missing)
self._delete(_share.Share, share, ignore_missing=ignore_missing)
def update_share(self, share_id, **attrs):
"""Updates details of a single share.
@ -172,12 +162,7 @@ class Proxy(proxy.Proxy):
res.revert_to_snapshot(self, snapshot_id)
def resize_share(
self,
share_id,
new_size,
no_shrink=False,
no_extend=False,
force=False
self, share_id, new_size, no_shrink=False, no_extend=False, force=False
):
"""Resizes a share, extending/shrinking the share as needed.
@ -206,8 +191,9 @@ class Proxy(proxy.Proxy):
elif new_size < res.size and no_shrink is not True:
res.shrink_share(self, new_size)
def wait_for_status(self, res, status='active', failures=None,
interval=2, wait=120):
def wait_for_status(
self, res, status='active', failures=None, interval=2, wait=120
):
"""Wait for a resource to be in a particular status.
:param res: The resource to wait on to reach the specified status.
The resource must have a ``status`` attribute.
@ -229,7 +215,8 @@ class Proxy(proxy.Proxy):
"""
failures = [] if failures is None else failures
return resource.wait_for_status(
self, res, status, failures, interval, wait)
self, res, status, failures, interval, wait
)
def storage_pools(self, details=True, **query):
"""Lists all back-end storage pools with details
@ -248,7 +235,8 @@ class Proxy(proxy.Proxy):
"""
base_path = '/scheduler-stats/pools/detail' if details else None
return self._list(
_storage_pool.StoragePool, base_path=base_path, **query)
_storage_pool.StoragePool, base_path=base_path, **query
)
def user_messages(self, **query):
"""List shared file system user messages
@ -278,8 +266,7 @@ class Proxy(proxy.Proxy):
:rtype:
:class:`~openstack.shared_file_system.v2.user_message.UserMessage`
"""
return self._list(
_user_message.UserMessage, **query)
return self._list(_user_message.UserMessage, **query)
def get_user_message(self, message_id):
"""List details of a single user message
@ -300,8 +287,10 @@ class Proxy(proxy.Proxy):
:class:`~openstack.shared_file_system.v2.user_message.UserMessage`
"""
return self._delete(
_user_message.UserMessage, message_id,
ignore_missing=ignore_missing)
_user_message.UserMessage,
message_id,
ignore_missing=ignore_missing,
)
def limits(self, **query):
"""Lists all share limits.
@ -312,8 +301,7 @@ class Proxy(proxy.Proxy):
:returns: A generator of manila share limits resources
:rtype: :class:`~openstack.shared_file_system.v2.limit.Limit`
"""
return self._list(
_limit.Limit, **query)
return self._list(_limit.Limit, **query)
def share_snapshots(self, details=True, **query):
"""Lists all share snapshots with details.
@ -329,7 +317,8 @@ class Proxy(proxy.Proxy):
"""
base_path = '/snapshots/detail' if details else None
return self._list(
_share_snapshot.ShareSnapshot, base_path=base_path, **query)
_share_snapshot.ShareSnapshot, base_path=base_path, **query
)
def get_share_snapshot(self, snapshot_id):
"""Lists details of a single share snapshot
@ -359,8 +348,9 @@ class Proxy(proxy.Proxy):
:rtype:
:class:`~openstack.shared_file_system.v2.share_snapshot.ShareSnapshot`
"""
return self._update(_share_snapshot.ShareSnapshot, snapshot_id,
**attrs)
return self._update(
_share_snapshot.ShareSnapshot, snapshot_id, **attrs
)
def delete_share_snapshot(self, snapshot_id, ignore_missing=True):
"""Deletes a single share snapshot
@ -369,8 +359,11 @@ class Proxy(proxy.Proxy):
:returns: Result of the ``delete``
:rtype: ``None``
"""
self._delete(_share_snapshot.ShareSnapshot, snapshot_id,
ignore_missing=ignore_missing)
self._delete(
_share_snapshot.ShareSnapshot,
snapshot_id,
ignore_missing=ignore_missing,
)
# ========= Network Subnets ==========
def share_network_subnets(self, share_network_id):
@ -384,10 +377,13 @@ class Proxy(proxy.Proxy):
"""
return self._list(
_share_network_subnet.ShareNetworkSubnet,
share_network_id=share_network_id)
share_network_id=share_network_id,
)
def get_share_network_subnet(
self, share_network_id, share_network_subnet_id,
self,
share_network_id,
share_network_subnet_id,
):
"""Lists details of a single share network subnet.
@ -403,7 +399,8 @@ class Proxy(proxy.Proxy):
return self._get(
_share_network_subnet.ShareNetworkSubnet,
share_network_subnet_id,
share_network_id=share_network_id)
share_network_id=share_network_id,
)
def create_share_network_subnet(self, share_network_id, **attrs):
"""Creates a share network subnet from attributes
@ -419,7 +416,8 @@ class Proxy(proxy.Proxy):
return self._create(
_share_network_subnet.ShareNetworkSubnet,
**attrs,
share_network_id=share_network_id)
share_network_id=share_network_id
)
def delete_share_network_subnet(
self, share_network_id, share_network_subnet, ignore_missing=True
@ -438,7 +436,8 @@ class Proxy(proxy.Proxy):
_share_network_subnet.ShareNetworkSubnet,
share_network_subnet,
share_network_id=share_network_id,
ignore_missing=ignore_missing)
ignore_missing=ignore_missing,
)
def wait_for_delete(self, res, interval=2, wait=120):
"""Wait for a resource to be deleted.
@ -474,8 +473,11 @@ class Proxy(proxy.Proxy):
share_snapshot_instance.ShareSnapshotInstance`
"""
base_path = '/snapshot-instances/detail' if details else None
return self._list(_share_snapshot_instance.ShareSnapshotInstance,
base_path=base_path, **query)
return self._list(
_share_snapshot_instance.ShareSnapshotInstance,
base_path=base_path,
**query
)
def get_share_snapshot_instance(self, snapshot_instance_id):
"""Lists details of a single share snapshot instance
@ -485,8 +487,10 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.shared_file_system.v2.
share_snapshot_instance.ShareSnapshotInstance`
"""
return self._get(_share_snapshot_instance.ShareSnapshotInstance,
snapshot_instance_id)
return self._get(
_share_snapshot_instance.ShareSnapshotInstance,
snapshot_instance_id,
)
def share_networks(self, details=True, **query):
"""Lists all share networks with details.
@ -508,7 +512,8 @@ class Proxy(proxy.Proxy):
"""
base_path = '/share-networks/detail' if details else None
return self._list(
_share_network.ShareNetwork, base_path=base_path, **query)
_share_network.ShareNetwork, base_path=base_path, **query
)
def get_share_network(self, share_network_id):
"""Lists details of a single share network
@ -527,8 +532,10 @@ class Proxy(proxy.Proxy):
:rtype: ``None``
"""
self._delete(
_share_network.ShareNetwork, share_network_id,
ignore_missing=ignore_missing)
_share_network.ShareNetwork,
share_network_id,
ignore_missing=ignore_missing,
)
def update_share_network(self, share_network_id, **attrs):
"""Updates details of a single share network.
@ -540,7 +547,8 @@ class Proxy(proxy.Proxy):
share_network.ShareNetwork`
"""
return self._update(
_share_network.ShareNetwork, share_network_id, **attrs)
_share_network.ShareNetwork, share_network_id, **attrs
)
def create_share_network(self, **attrs):
"""Creates a share network from attributes
@ -570,8 +578,7 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.shared_file_system.v2.
share_instance.ShareInstance`
"""
return self._list(
_share_instance.ShareInstance, **query)
return self._list(_share_instance.ShareInstance, **query)
def get_share_instance(self, share_instance_id):
"""Shows details for a single share instance
@ -592,8 +599,9 @@ class Proxy(proxy.Proxy):
:returns: ``None``
"""
res = self._get_resource(_share_instance.ShareInstance,
share_instance_id)
res = self._get_resource(
_share_instance.ShareInstance, share_instance_id
)
res.reset_status(self, status)
def delete_share_instance(self, share_instance_id):
@ -603,8 +611,9 @@ class Proxy(proxy.Proxy):
:returns: ``None``
"""
res = self._get_resource(_share_instance.ShareInstance,
share_instance_id)
res = self._get_resource(
_share_instance.ShareInstance, share_instance_id
)
res.force_delete(self)
def export_locations(self, share_id):
@ -615,8 +624,9 @@ class Proxy(proxy.Proxy):
:rtype: List of :class:`~openstack.shared_filesystem_storage.v2.
share_export_locations.ShareExportLocations`
"""
return self._list(_share_export_locations.ShareExportLocation,
share_id=share_id)
return self._list(
_share_export_locations.ShareExportLocation, share_id=share_id
)
def get_export_location(self, export_location, share_id):
"""List details of export location
@ -631,7 +641,9 @@ class Proxy(proxy.Proxy):
export_location_id = resource.Resource._get_id(export_location)
return self._get(
_share_export_locations.ShareExportLocation,
export_location_id, share_id=share_id)
export_location_id,
share_id=share_id,
)
def access_rules(self, share, **query):
"""Lists the access rules on a share.
@ -642,8 +654,8 @@ class Proxy(proxy.Proxy):
"""
share = self._get_resource(_share.Share, share)
return self._list(
_share_access_rule.ShareAccessRule,
share_id=share.id, **query)
_share_access_rule.ShareAccessRule, share_id=share.id, **query
)
def get_access_rule(self, access_id):
"""List details of an access rule.
@ -653,8 +665,7 @@ class Proxy(proxy.Proxy):
:rtype: :class:`~openstack.shared_file_system.v2.
share_access_rules.ShareAccessRules`
"""
return self._get(
_share_access_rule.ShareAccessRule, access_id)
return self._get(_share_access_rule.ShareAccessRule, access_id)
def create_access_rule(self, share_id, **attrs):
"""Creates an access rule from attributes
@ -670,7 +681,8 @@ class Proxy(proxy.Proxy):
"""
base_path = "/shares/%s/action" % (share_id,)
return self._create(
_share_access_rule.ShareAccessRule, base_path=base_path, **attrs)
_share_access_rule.ShareAccessRule, base_path=base_path, **attrs
)
def delete_access_rule(self, access_id, share_id, ignore_missing=True):
"""Deletes an access rule
@ -680,6 +692,5 @@ class Proxy(proxy.Proxy):
:rtype: ``None``
"""
res = self._get_resource(
_share_access_rule.ShareAccessRule, access_id)
res = self._get_resource(_share_access_rule.ShareAccessRule, access_id)
res.delete(self, share_id, ignore_missing=ignore_missing)

39
openstack/shared_file_system/v2/limit.py
View File

@ -29,47 +29,46 @@ class Limit(resource.Resource):
#: The maximum number of replica gigabytes that are allowed
#: in a project.
maxTotalReplicaGigabytes = resource.Body(
"maxTotalReplicaGigabytes", type=int)
"maxTotalReplicaGigabytes", type=int
)
#: The total maximum number of shares that are allowed in a project.
maxTotalShares = resource.Body("maxTotalShares", type=int)
#: The total maximum number of share gigabytes that are allowed in a
#: project.
maxTotalShareGigabytes = resource.Body(
"maxTotalShareGigabytes", type=int)
maxTotalShareGigabytes = resource.Body("maxTotalShareGigabytes", type=int)
#: The total maximum number of share-networks that are allowed in a
#: project.
maxTotalShareNetworks = resource.Body(
"maxTotalShareNetworks", type=int)
maxTotalShareNetworks = resource.Body("maxTotalShareNetworks", type=int)
#: The total maximum number of share snapshots that are allowed in a
#: project.
maxTotalShareSnapshots = resource.Body(
"maxTotalShareSnapshots", type=int)
maxTotalShareSnapshots = resource.Body("maxTotalShareSnapshots", type=int)
#: The maximum number of share replicas that is allowed.
maxTotalShareReplicas = resource.Body(
"maxTotalShareReplicas", type=int)
maxTotalShareReplicas = resource.Body("maxTotalShareReplicas", type=int)
#: The total maximum number of snapshot gigabytes that are allowed
#: in a project.
maxTotalSnapshotGigabytes = resource.Body(
"maxTotalSnapshotGigabytes", type=int)
"maxTotalSnapshotGigabytes", type=int
)
#: The total number of replica gigabytes used in a project by
#: share replicas.
totalReplicaGigabytesUsed = resource.Body(
"totalReplicaGigabytesUsed", type=int)
"totalReplicaGigabytesUsed", type=int
)
#: The total number of gigabytes used in a project by shares.
totalShareGigabytesUsed = resource.Body(
"totalShareGigabytesUsed", type=int)
"totalShareGigabytesUsed", type=int
)
#: The total number of created shares in a project.
totalSharesUsed = resource.Body(
"totalSharesUsed", type=int)
totalSharesUsed = resource.Body("totalSharesUsed", type=int)
#: The total number of created share-networks in a project.
totalShareNetworksUsed = resource.Body(
"totalShareNetworksUsed", type=int)
totalShareNetworksUsed = resource.Body("totalShareNetworksUsed", type=int)
#: The total number of created share snapshots in a project.
totalShareSnapshotsUsed = resource.Body(
"totalShareSnapshotsUsed", type=int)
"totalShareSnapshotsUsed", type=int
)
#: The total number of gigabytes used in a project by snapshots.
totalSnapshotGigabytesUsed = resource.Body(
"totalSnapshotGigabytesUsed", type=int)
"totalSnapshotGigabytesUsed", type=int
)
#: The total number of created share replicas in a project.
totalShareReplicasUsed = resource.Body(
"totalShareReplicasUsed", type=int)
totalShareReplicasUsed = resource.Body("totalShareReplicasUsed", type=int)

24
openstack/shared_file_system/v2/share.py
View File

@ -46,18 +46,20 @@ class Share(resource.Resource):
#: Whether or not this share supports snapshots that can be
#: cloned into new shares.
is_creating_new_share_from_snapshot_supported = resource.Body(
"create_share_from_snapshot_support", type=bool)
"create_share_from_snapshot_support", type=bool
)
#: Whether the share's snapshots can be mounted directly and access
#: controlled independently or not.
is_mounting_snapshot_supported = resource.Body(
"mount_snapshot_support", type=bool)
"mount_snapshot_support", type=bool
)
#: Whether the share can be reverted to its latest snapshot or not.
is_reverting_to_snapshot_supported = resource.Body(
"revert_to_snapshot_support", type=bool)
"revert_to_snapshot_support", type=bool
)
#: An extra specification that filters back ends by whether the share
#: supports snapshots or not.
is_snapshot_supported = resource.Body(
"snapshot_support", type=bool)
is_snapshot_supported = resource.Body("snapshot_support", type=bool)
#: Indicates whether the share has replicas or not.
is_replicated = resource.Body("has_replicas", type=bool)
#: One or more metadata key and value pairs as a dictionary of strings.
@ -91,7 +93,8 @@ class Share(resource.Resource):
#: The ID of the group snapshot instance that was used to create
#: this share.
source_share_group_snapshot_member_id = resource.Body(
"source_share_group_snapshot_member_id", type=str)
"source_share_group_snapshot_member_id", type=str
)
#: The share status
status = resource.Body("status", type=str)
#: For the share migration, the migration task state.
@ -109,14 +112,11 @@ class Share(resource.Resource):
headers = {'Accept': ''}
if microversion is None:
microversion = \
self._get_microversion(session, action=action)
microversion = self._get_microversion(session, action=action)
response = session.post(
url,
json=body,
headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)
exceptions.raise_from_response(response)
return response

24
openstack/shared_file_system/v2/share_access_rule.py
View File

@ -55,28 +55,26 @@ class ShareAccessRule(resource.Resource):
#: the service’s database.
updated_at = resource.Body("updated_at", type=str)
def _action(self, session, body, url,
action='patch', microversion=None):
def _action(self, session, body, url, action='patch', microversion=None):
headers = {'Accept': ''}
if microversion is None:
microversion = \
self._get_microversion(session, action=action)
microversion = self._get_microversion(session, action=action)
session.post(
url,
json=body,
headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)
def create(self, session, **kwargs):
return super().create(session,
resource_request_key='allow_access',
resource_response_key='access',
**kwargs)
return super().create(
session,
resource_request_key='allow_access',
resource_response_key='access',
**kwargs
)
def delete(self, session, share_id, ignore_missing=True):
body = {'deny_access': {'access_id' : self.id}}
body = {'deny_access': {'access_id': self.id}}
url = utils.urljoin('/shares', share_id, 'action')
try:
response = self._action(session, body, url)

5
openstack/shared_file_system/v2/share_instance.py
View File

@ -66,8 +66,9 @@ class ShareInstance(resource.Resource):
# Set microversion override
extra_attrs['microversion'] = microversion
else:
extra_attrs['microversion'] = \
self._get_microversion(session, action=action)
extra_attrs['microversion'] = self._get_microversion(
session, action=action
)
response = session.post(url, json=body, headers=headers, **extra_attrs)
exceptions.raise_from_response(response)
return response

12
openstack/shared_file_system/v2/share_network.py
View File

@ -27,9 +27,15 @@ class ShareNetwork(resource.Resource):
allow_head = False
_query_mapping = resource.QueryParameters(
"project_id", "name", "description",
"created_since", "created_before", "security_service_id",
"limit", "offset", all_projects="all_tenants",
"project_id",
"name",
"description",
"created_since",
"created_before",
"security_service_id",
"limit",
"offset",
all_projects="all_tenants",
)
#: Properties

7
openstack/shared_file_system/v2/share_network_subnet.py
View File

@ -56,7 +56,6 @@ class ShareNetworkSubnet(resource.Resource):
updated_at = resource.Body("updated_at", type=str)
def create(self, session, **kwargs):
return super().\
create(session,
resource_request_key='share-network-subnet',
**kwargs)
return super().create(
session, resource_request_key='share-network-subnet', **kwargs
)

7
openstack/shared_file_system/v2/share_snapshot.py
View File

@ -26,9 +26,7 @@ class ShareSnapshot(resource.Resource):
allow_list = True
allow_head = False
_query_mapping = resource.QueryParameters(
"snapshot_id"
)
_query_mapping = resource.QueryParameters("snapshot_id")
#: Properties
#: The date and time stamp when the resource was
@ -39,8 +37,7 @@ class ShareSnapshot(resource.Resource):
#: The user defined name of the resource.
display_name = resource.Body("display_name", type=str)
#: The user defined description of the resource
display_description = resource.Body(
"display_description", type=str)
display_description = resource.Body("display_description", type=str)
#: ID of the project that the snapshot belongs to.
project_id = resource.Body("project_id", type=str)
#: The UUID of the source share that was used to

6
openstack/shared_file_system/v2/storage_pool.py
View File

@ -27,7 +27,11 @@ class StoragePool(resource.Resource):
allow_head = False
_query_mapping = resource.QueryParameters(
'pool', 'backend', 'host', 'capabilities', 'share_type',
'pool',
'backend',
'host',
'capabilities',
'share_type',
)
#: Properties

4
openstack/shared_file_system/v2/user_message.py
View File

@ -25,9 +25,7 @@ class UserMessage(resource.Resource):
allow_list = True
allow_head = False
_query_mapping = resource.QueryParameters(
"message_id"
)
_query_mapping = resource.QueryParameters("message_id")
_max_microversion = '2.37'

39
openstack/tests/functional/shared_file_system/base.py
View File

@ -20,40 +20,49 @@ class BaseSharedFileSystemTest(base.BaseFunctionalTest):
def setUp(self):
super(BaseSharedFileSystemTest, self).setUp()
self.require_service('shared-file-system',
min_microversion=self.min_microversion)
self.require_service(
'shared-file-system', min_microversion=self.min_microversion
)
self._set_operator_cloud(shared_file_system_api_version='2.63')
self._set_user_cloud(shared_file_system_api_version='2.63')
def create_share(self, **kwargs):
share = self.user_cloud.share.create_share(**kwargs)
self.addCleanup(self.user_cloud.share.delete_share,
share.id,
ignore_missing=True)
self.addCleanup(
self.user_cloud.share.delete_share, share.id, ignore_missing=True
)
self.user_cloud.share.wait_for_status(
share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(share.id)
return share
def create_share_snapshot(self, share_id, **kwargs):
share_snapshot = self.user_cloud.share.create_share_snapshot(
share_id=share_id, force=True)
self.addCleanup(resource.wait_for_delete,
self.user_cloud.share, share_snapshot,
wait=self._wait_for_timeout,
interval=2)
self.addCleanup(self.user_cloud.share.delete_share_snapshot,
share_snapshot.id,
ignore_missing=False)
share_id=share_id, force=True
)
self.addCleanup(
resource.wait_for_delete,
self.user_cloud.share,
share_snapshot,
wait=self._wait_for_timeout,
interval=2,
)
self.addCleanup(
self.user_cloud.share.delete_share_snapshot,
share_snapshot.id,
ignore_missing=False,
)
self.user_cloud.share.wait_for_status(
share_snapshot,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(share_snapshot.id)
return share_snapshot

16
openstack/tests/functional/shared_file_system/test_export_locations.py
View File

@ -22,8 +22,12 @@ class TestExportLocation(base.BaseSharedFileSystemTest):
self.SHARE_NAME = self.getUniqueString()
my_share = self.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.SHARE_ID = my_share.id
def test_export_locations(self):
@ -33,8 +37,12 @@ class TestExportLocation(base.BaseSharedFileSystemTest):
self.assertGreater(len(list(exs)), 0)
for ex in exs:
for attribute in (
'id', 'path', 'share_instance_id',
'updated_at', 'created_at'):
'id',
'path',
'share_instance_id',
'updated_at',
'created_at',
):
self.assertTrue(hasattr(ex, attribute))
self.assertIsInstance(getattr(ex, attribute), 'str')
for attribute in ('is_preferred', 'is_admin'):

31
openstack/tests/functional/shared_file_system/test_limit.py
View File

@ -14,23 +14,24 @@ from openstack.tests.functional.shared_file_system import base
class LimitTest(base.BaseSharedFileSystemTest):
def test_limits(self):
limits = self.user_cloud.shared_file_system.limits()
self.assertGreater(len(list(limits)), 0)
for limit in limits:
for attribute in ("maxTotalReplicaGigabytes",
"maxTotalShares",
"maxTotalShareGigabytes",
"maxTotalShareNetworks",
"maxTotalShareSnapshots",
"maxTotalShareReplicas",
"maxTotalSnapshotGigabytes",
"totalReplicaGigabytesUsed",
"totalShareGigabytesUsed",
"totalSharesUsed",
"totalShareNetworksUsed",
"totalShareSnapshotsUsed",
"totalSnapshotGigabytesUsed",
"totalShareReplicasUsed"):
for attribute in (
"maxTotalReplicaGigabytes",
"maxTotalShares",
"maxTotalShareGigabytes",
"maxTotalShareNetworks",
"maxTotalShareSnapshots",
"maxTotalShareReplicas",
"maxTotalSnapshotGigabytes",
"totalReplicaGigabytesUsed",
"totalShareGigabytesUsed",
"totalSharesUsed",
"totalShareNetworksUsed",
"totalShareSnapshotsUsed",
"totalSnapshotGigabytesUsed",
"totalShareReplicasUsed",
):
self.assertTrue(hasattr(limit, attribute))

73
openstack/tests/functional/shared_file_system/test_share.py
View File

@ -15,19 +15,20 @@ from openstack.tests.functional.shared_file_system import base
class ShareTest(base.BaseSharedFileSystemTest):
def setUp(self):
super(ShareTest, self).setUp()
self.SHARE_NAME = self.getUniqueString()
my_share = self.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.SHARE_ID = my_share.id
self.SHARE_SIZE = my_share.size
my_share_snapshot = self.create_share_snapshot(
share_id=self.SHARE_ID
)
my_share_snapshot = self.create_share_snapshot(share_id=self.SHARE_ID)
self.SHARE_SNAPSHOT_ID = my_share_snapshot.id
def test_get(self):
@ -44,73 +45,73 @@ class ShareTest(base.BaseSharedFileSystemTest):
def test_update(self):
updated_share = self.user_cloud.share.update_share(
self.SHARE_ID, display_description='updated share')
get_updated_share = self.user_cloud.share.get_share(
updated_share.id)
self.SHARE_ID, display_description='updated share'
)
get_updated_share = self.user_cloud.share.get_share(updated_share.id)
self.assertEqual('updated share', get_updated_share.description)
def test_revert_share_to_snapshot(self):
self.user_cloud.share.revert_share_to_snapshot(
self.SHARE_ID, self.SHARE_SNAPSHOT_ID)
get_reverted_share = self.user_cloud.share.get_share(
self.SHARE_ID)
self.SHARE_ID, self.SHARE_SNAPSHOT_ID
)
get_reverted_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_reverted_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(get_reverted_share.id)
def test_resize_share_larger(self):
larger_size = 3
self.user_cloud.share.resize_share(
self.SHARE_ID, larger_size)
self.user_cloud.share.resize_share(self.SHARE_ID, larger_size)
get_resized_share = self.user_cloud.share.get_share(
self.SHARE_ID)
get_resized_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_resized_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertEqual(larger_size, get_resized_share.size)
def test_resize_share_smaller(self):
# Resize to 3 GiB
smaller_size = 1
self.user_cloud.share.resize_share(
self.SHARE_ID, smaller_size)
self.user_cloud.share.resize_share(self.SHARE_ID, smaller_size)
get_resized_share = self.user_cloud.share.get_share(
self.SHARE_ID)
get_resized_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_resized_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertEqual(smaller_size, get_resized_share.size)
def test_resize_share_larger_no_extend(self):
larger_size = 3
self.user_cloud.share.resize_share(
self.SHARE_ID, larger_size, no_extend=True)
self.SHARE_ID, larger_size, no_extend=True
)
get_resized_share = self.user_cloud.share.get_share(
self.SHARE_ID)
get_resized_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_resized_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
# Assert that no change was made.
self.assertEqual(self.SHARE_SIZE, get_resized_share.size)
@ -119,17 +120,18 @@ class ShareTest(base.BaseSharedFileSystemTest):
smaller_size = 1
self.user_cloud.share.resize_share(
self.SHARE_ID, smaller_size, no_shrink=True)
self.SHARE_ID, smaller_size, no_shrink=True
)
get_resized_share = self.user_cloud.share.get_share(
self.SHARE_ID)
get_resized_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_resized_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
# Assert that no change was made.
self.assertEqual(self.SHARE_SIZE, get_resized_share.size)
@ -138,15 +140,16 @@ class ShareTest(base.BaseSharedFileSystemTest):
# Resize to 3 GiB
larger_size = 3
self.user_cloud.share.resize_share(
self.SHARE_ID, larger_size, force=True)
self.SHARE_ID, larger_size, force=True
)
get_resized_share = self.user_cloud.share.get_share(
self.SHARE_ID)
get_resized_share = self.user_cloud.share.get_share(self.SHARE_ID)
self.user_cloud.share.wait_for_status(
get_resized_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertEqual(larger_size, get_resized_share.size)

35
openstack/tests/functional/shared_file_system/test_share_access_rule.py
View File

@ -14,20 +14,24 @@ from openstack.tests.functional.shared_file_system import base
class ShareAccessRuleTest(base.BaseSharedFileSystemTest):
def setUp(self):
super(ShareAccessRuleTest, self).setUp()
self.SHARE_NAME = self.getUniqueString()
mys = self.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.user_cloud.shared_file_system.wait_for_status(
mys,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(mys)
self.assertIsNotNone(mys.id)
self.SHARE_ID = mys.id
@ -36,16 +40,14 @@ class ShareAccessRuleTest(base.BaseSharedFileSystemTest):
self.SHARE_ID,
access_level="rw",
access_type="ip",
access_to="0.0.0.0/0"
access_to="0.0.0.0/0",
)
self.ACCESS_ID = access_rule.id
self.RESOURCE_KEY = access_rule.resource_key
def tearDown(self):
acr = self.user_cloud.share.delete_access_rule(
self.ACCESS_ID,
self.SHARE_ID,
ignore_missing=True
self.ACCESS_ID, self.SHARE_ID, ignore_missing=True
)
self.assertIsNone(acr)
@ -59,12 +61,19 @@ class ShareAccessRuleTest(base.BaseSharedFileSystemTest):
def test_list_access_rules(self):
rules = self.user_cloud.shared_file_system.access_rules(
self.SHARE,
details=True
self.SHARE, details=True
)
self.assertGreater(len(list(rules)), 0)
for rule in rules:
for attribute in ('id', 'created_at', 'updated_at',
'access_level', 'access_type', 'access_to',
'share_id', 'access_key', 'metadata'):
for attribute in (
'id',
'created_at',
'updated_at',
'access_level',
'access_type',
'access_to',
'share_id',
'access_key',
'metadata',
):
self.assertTrue(hasattr(rule, attribute))

42
openstack/tests/functional/shared_file_system/test_share_instance.py
View File

@ -25,8 +25,12 @@ class ShareInstanceTest(base.BaseSharedFileSystemTest):
self.SHARE_NAME = self.getUniqueString()
my_share = self.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.SHARE_ID = my_share.id
instances_list = self.operator_cloud.share.share_instances()
self.SHARE_INSTANCE_ID = None
@ -36,7 +40,8 @@ class ShareInstanceTest(base.BaseSharedFileSystemTest):
def test_get(self):
sot = self.operator_cloud.share.get_share_instance(
self.SHARE_INSTANCE_ID)
self.SHARE_INSTANCE_ID
)
assert isinstance(sot, _share_instance.ShareInstance)
self.assertEqual(self.SHARE_INSTANCE_ID, sot.id)
@ -44,25 +49,36 @@ class ShareInstanceTest(base.BaseSharedFileSystemTest):
share_instances = self.operator_cloud.share.share_instances()
self.assertGreater(len(list(share_instances)), 0)
for share_instance in share_instances:
for attribute in ('id', 'name', 'created_at',
'access_rules_status',
'availability_zone'):
for attribute in (
'id',
'name',
'created_at',
'access_rules_status',
'availability_zone',
):
self.assertTrue(hasattr(share_instance, attribute))
def test_reset(self):
res = self.operator_cloud.share.reset_share_instance_status(
self.SHARE_INSTANCE_ID, 'error')
self.SHARE_INSTANCE_ID, 'error'
)
self.assertIsNone(res)
sot = self.operator_cloud.share.get_share_instance(
self.SHARE_INSTANCE_ID)
self.SHARE_INSTANCE_ID
)
self.assertEqual('error', sot.status)
def test_delete(self):
sot = self.operator_cloud.share.get_share_instance(
self.SHARE_INSTANCE_ID)
self.SHARE_INSTANCE_ID
)
fdel = self.operator_cloud.share.delete_share_instance(
self.SHARE_INSTANCE_ID)
resource.wait_for_delete(self.operator_cloud.share, sot,
wait=self._wait_for_timeout,
interval=2)
self.SHARE_INSTANCE_ID
)
resource.wait_for_delete(
self.operator_cloud.share,
sot,
wait=self._wait_for_timeout,
interval=2,
)
self.assertIsNone(fdel)

20
openstack/tests/functional/shared_file_system/test_share_network.py
View File

@ -15,27 +15,28 @@ from openstack.tests.functional.shared_file_system import base
class ShareNetworkTest(base.BaseSharedFileSystemTest):
def setUp(self):
super(ShareNetworkTest, self).setUp()
self.SHARE_NETWORK_NAME = self.getUniqueString()
snt = self.user_cloud.shared_file_system.create_share_network(
name=self.SHARE_NETWORK_NAME)
name=self.SHARE_NETWORK_NAME
)
self.assertIsNotNone(snt)
self.assertIsNotNone(snt.id)
self.SHARE_NETWORK_ID = snt.id
def tearDown(self):
sot = self.user_cloud.shared_file_system.delete_share_network(
self.SHARE_NETWORK_ID,
ignore_missing=True)
self.SHARE_NETWORK_ID, ignore_missing=True
)
self.assertIsNone(sot)
super(ShareNetworkTest, self).tearDown()
def test_get(self):
sot = self.user_cloud.shared_file_system.get_share_network(
self.SHARE_NETWORK_ID)
self.SHARE_NETWORK_ID
)
assert isinstance(sot, _share_network.ShareNetwork)
self.assertEqual(self.SHARE_NETWORK_ID, sot.id)
@ -50,12 +51,13 @@ class ShareNetworkTest(base.BaseSharedFileSystemTest):
def test_delete_share_network(self):
sot = self.user_cloud.shared_file_system.delete_share_network(
self.SHARE_NETWORK_ID)
self.SHARE_NETWORK_ID
)
self.assertIsNone(sot)
def test_update(self):
unt = self.user_cloud.shared_file_system.update_share_network(
self.SHARE_NETWORK_ID, description='updated share network')
get_unt = self.user_cloud.shared_file_system.get_share_network(
unt.id)
self.SHARE_NETWORK_ID, description='updated share network'
)
get_unt = self.user_cloud.shared_file_system.get_share_network(unt.id)
self.assertEqual('updated share network', get_unt.description)

71
openstack/tests/functional/shared_file_system/test_share_network_subnet.py
View File

@ -11,65 +11,76 @@
# under the License.
from openstack.shared_file_system.v2 import (
share_network_subnet as _share_network_subnet
share_network_subnet as _share_network_subnet,
)
from openstack.tests.functional.shared_file_system import base
class ShareNetworkSubnetTest(base.BaseSharedFileSystemTest):
def setUp(self):
super().setUp()
zones = self.user_cloud.shared_file_system.\
availability_zones()
zones = self.user_cloud.shared_file_system.availability_zones()
first_zone = next(zones)
self.SHARE_NETWORK_NAME = self.getUniqueString()
snt = self.user_cloud.shared_file_system.create_share_network(
name=self.SHARE_NETWORK_NAME)
name=self.SHARE_NETWORK_NAME
)
self.assertIsNotNone(snt)
self.assertIsNotNone(snt.id)
self.SHARE_NETWORK_ID = snt.id
snsb = self.user_cloud.shared_file_system.\
create_share_network_subnet(self.SHARE_NETWORK_ID,
availability_zone=first_zone.name)
snsb = self.user_cloud.shared_file_system.create_share_network_subnet(
self.SHARE_NETWORK_ID, availability_zone=first_zone.name
)
self.assertIsNotNone(snsb)
self.assertIsNotNone(snsb.id)
self.SHARE_NETWORK_SUBNET_ID = snsb.id
def tearDown(self):
subnet = self.user_cloud.shared_file_system.\
get_share_network_subnet(self.SHARE_NETWORK_ID,
self.SHARE_NETWORK_SUBNET_ID)
fdel = self.user_cloud.shared_file_system.\
delete_share_network_subnet(self.SHARE_NETWORK_ID,
self.SHARE_NETWORK_SUBNET_ID,
ignore_missing=True)
subnet = self.user_cloud.shared_file_system.get_share_network_subnet(
self.SHARE_NETWORK_ID, self.SHARE_NETWORK_SUBNET_ID
)
fdel = self.user_cloud.shared_file_system.delete_share_network_subnet(
self.SHARE_NETWORK_ID,
self.SHARE_NETWORK_SUBNET_ID,
ignore_missing=True,
)
self.assertIsNone(fdel)
self.user_cloud.shared_file_system.\
wait_for_delete(subnet)
sot = self.user_cloud.shared_file_system.\
delete_share_network(self.SHARE_NETWORK_ID,
ignore_missing=True)
self.user_cloud.shared_file_system.wait_for_delete(subnet)
sot = self.user_cloud.shared_file_system.delete_share_network(
self.SHARE_NETWORK_ID, ignore_missing=True
)
self.assertIsNone(sot)
super().tearDown()
def test_get(self):
sub = self.user_cloud.shared_file_system.\
get_share_network_subnet(self.SHARE_NETWORK_ID,
self.SHARE_NETWORK_SUBNET_ID)
sub = self.user_cloud.shared_file_system.get_share_network_subnet(
self.SHARE_NETWORK_ID, self.SHARE_NETWORK_SUBNET_ID
)
assert isinstance(sub, _share_network_subnet.ShareNetworkSubnet)
def test_list(self):
subs = self.user_cloud.shared_file_system.share_network_subnets(
self.SHARE_NETWORK_ID)
self.SHARE_NETWORK_ID
)
self.assertGreater(len(list(subs)), 0)
for sub in subs:
for attribute in ('id', 'name', 'created_at', 'updated_at',
'share_network_id', 'availability_zone',
'cidr', 'gateway', 'ip_version', 'mtu',
'network_type', 'neutron_net_id',
'neutron_subnet_id', 'segmentation_id',
'share_network_name'):
for attribute in (
'id',
'name',
'created_at',
'updated_at',
'share_network_id',
'availability_zone',
'cidr',
'gateway',
'ip_version',
'mtu',
'network_type',
'neutron_net_id',
'neutron_subnet_id',
'segmentation_id',
'share_network_name',
):
self.assertTrue(hasattr(sub, attribute))

50
openstack/tests/functional/shared_file_system/test_share_snapshot.py
View File

@ -14,48 +14,56 @@ from openstack.tests.functional.shared_file_system import base
class ShareSnapshotTest(base.BaseSharedFileSystemTest):
def setUp(self):
super(ShareSnapshotTest, self).setUp()
self.SHARE_NAME = self.getUniqueString()
self.SNAPSHOT_NAME = self.getUniqueString()
my_share = self.operator_cloud.shared_file_system.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.operator_cloud.shared_file_system.wait_for_status(
my_share,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(my_share)
self.assertIsNotNone(my_share.id)
self.SHARE_ID = my_share.id
msp = self.operator_cloud.shared_file_system.create_share_snapshot(
share_id=self.SHARE_ID, name=self.SNAPSHOT_NAME, force=True)
share_id=self.SHARE_ID, name=self.SNAPSHOT_NAME, force=True
)
self.operator_cloud.shared_file_system.wait_for_status(
msp,
status='available',
failures=['error'],
interval=5,
wait=self._wait_for_timeout)
wait=self._wait_for_timeout,
)
self.assertIsNotNone(msp.id)
self.SNAPSHOT_ID = msp.id
def tearDown(self):
snpt = self.operator_cloud.shared_file_system.get_share_snapshot(
self.SNAPSHOT_ID)
self.SNAPSHOT_ID
)
sot = self.operator_cloud.shared_file_system.delete_share_snapshot(
snpt, ignore_missing=False
)
self.operator_cloud.shared_file_system.wait_for_delete(
snpt, interval=2, wait=self._wait_for_timeout)
snpt, interval=2, wait=self._wait_for_timeout
)
self.assertIsNone(sot)
sot = self.operator_cloud.shared_file_system.delete_share(
self.SHARE_ID,
ignore_missing=False)
self.SHARE_ID, ignore_missing=False
)
self.assertIsNone(sot)
super(ShareSnapshotTest, self).tearDown()
@ -71,14 +79,26 @@ class ShareSnapshotTest(base.BaseSharedFileSystemTest):
)
self.assertGreater(len(list(snaps)), 0)
for snap in snaps:
for attribute in ('id', 'name', 'created_at', 'updated_at',
'description', 'share_id', 'share_proto',
'share_size', 'size', 'status', 'user_id'):
for attribute in (
'id',
'name',
'created_at',
'updated_at',
'description',
'share_id',
'share_proto',
'share_size',
'size',
'status',
'user_id',
):
self.assertTrue(hasattr(snap, attribute))
def test_update(self):
u_snap = self.operator_cloud.shared_file_system.update_share_snapshot(
self.SNAPSHOT_ID, display_description='updated share snapshot')
self.SNAPSHOT_ID, display_description='updated share snapshot'
)
get_u_snap = self.operator_cloud.shared_file_system.get_share_snapshot(
u_snap.id)
u_snap.id
)
self.assertEqual('updated share snapshot', get_u_snap.description)

16
openstack/tests/functional/shared_file_system/test_share_snapshot_instance.py
View File

@ -14,22 +14,24 @@ from openstack.tests.functional.shared_file_system import base
class ShareSnapshotInstanceTest(base.BaseSharedFileSystemTest):
def setUp(self):
super(ShareSnapshotInstanceTest, self).setUp()
self.SHARE_NAME = self.getUniqueString()
my_share = self.create_share(
name=self.SHARE_NAME, size=2, share_type="dhss_false",
share_protocol='NFS', description=None)
self.SHARE_ID = my_share.id
self.create_share_snapshot(
share_id=self.SHARE_ID
name=self.SHARE_NAME,
size=2,
share_type="dhss_false",
share_protocol='NFS',
description=None,
)
self.SHARE_ID = my_share.id
self.create_share_snapshot(share_id=self.SHARE_ID)
def test_share_snapshot_instances(self):
sots = \
sots = (
self.operator_cloud.shared_file_system.share_snapshot_instances()
)
self.assertGreater(len(list(sots)), 0)
for sot in sots:
for attribute in ('id', 'name', 'created_at', 'updated_at'):

10
openstack/tests/functional/shared_file_system/test_storage_pool.py
View File

@ -14,11 +14,15 @@ from openstack.tests.functional.shared_file_system import base
class StoragePoolTest(base.BaseSharedFileSystemTest):
def test_storage_pools(self):
pools = self.operator_cloud.shared_file_system.storage_pools()
self.assertGreater(len(list(pools)), 0)
for pool in pools:
for attribute in ('pool', 'name', 'host', 'backend',
'capabilities'):
for attribute in (
'pool',
'name',
'host',
'backend',
'capabilities',
):
self.assertTrue(hasattr(pool, attribute))

19
openstack/tests/functional/shared_file_system/test_user_message.py
View File

@ -14,7 +14,6 @@ from openstack.tests.functional.shared_file_system import base
class UserMessageTest(base.BaseSharedFileSystemTest):
def test_user_messages(self):
# TODO(kafilat): We must intentionally cause an asynchronous failure to
# ensure that at least one user message exists;
@ -22,11 +21,19 @@ class UserMessageTest(base.BaseSharedFileSystemTest):
# self.assertGreater(len(list(u_messages)), 0)
for u_message in u_messages:
for attribute in (
'id', 'created_at', 'action_id', 'detail_id',
'expires_at', 'message_level', 'project_id', 'request_id',
'resource_id', 'resource_type', 'user_message'):
'id',
'created_at',
'action_id',
'detail_id',
'expires_at',
'message_level',
'project_id',
'request_id',
'resource_id',
'resource_type',
'user_message',
):
self.assertTrue(hasattr(u_message, attribute))
self.assertIsInstance(getattr(u_message, attribute), str)
self.conn.shared_file_system.delete_user_message(
u_message)
self.conn.shared_file_system.delete_user_message(u_message)

1
openstack/tests/unit/shared_file_system/v2/test_availability_zone.py
View File

@ -23,7 +23,6 @@ EXAMPLE = {
class TestAvailabilityZone(base.TestCase):
def test_basic(self):
az_resource = az.AvailabilityZone()
self.assertEqual('availability_zones', az_resource.resources_key)

73
openstack/tests/unit/shared_file_system/v2/test_limit.py
View File

@ -27,12 +27,11 @@ EXAMPLE = {
"maxTotalShareReplicas": 100,
"maxTotalReplicaGigabytes": 1000,
"totalShareReplicasUsed": 0,
"totalReplicaGigabytesUsed": 0
"totalReplicaGigabytesUsed": 0,
}
class TestLimit(base.TestCase):
def test_basic(self):
limits = limit.Limit()
self.assertEqual('limits', limits.resources_key)
@ -46,31 +45,45 @@ class TestLimit(base.TestCase):
def test_make_limits(self):
limits = limit.Limit(**EXAMPLE)
self.assertEqual(EXAMPLE['totalShareNetworksUsed'],
limits.totalShareNetworksUsed)
self.assertEqual(EXAMPLE['maxTotalShareGigabytes'],
limits.maxTotalShareGigabytes)
self.assertEqual(EXAMPLE['maxTotalShareNetworks'],
limits.maxTotalShareNetworks)
self.assertEqual(EXAMPLE['totalSharesUsed'],
limits.totalSharesUsed)
self.assertEqual(EXAMPLE['totalShareGigabytesUsed'],
limits.totalShareGigabytesUsed)
self.assertEqual(EXAMPLE['totalShareSnapshotsUsed'],
limits.totalShareSnapshotsUsed)
self.assertEqual(EXAMPLE['maxTotalShares'],
limits.maxTotalShares)
self.assertEqual(EXAMPLE['totalSnapshotGigabytesUsed'],
limits.totalSnapshotGigabytesUsed)
self.assertEqual(EXAMPLE['maxTotalSnapshotGigabytes'],
limits.maxTotalSnapshotGigabytes)
self.assertEqual(EXAMPLE['maxTotalShareSnapshots'],
limits.maxTotalShareSnapshots)
self.assertEqual(EXAMPLE['maxTotalShareReplicas'],
limits.maxTotalShareReplicas)
self.assertEqual(EXAMPLE['maxTotalReplicaGigabytes'],
limits.maxTotalReplicaGigabytes)
self.assertEqual(EXAMPLE['totalShareReplicasUsed'],
limits.totalShareReplicasUsed)
self.assertEqual(EXAMPLE['totalReplicaGigabytesUsed'],
limits.totalReplicaGigabytesUsed)
self.assertEqual(
EXAMPLE['totalShareNetworksUsed'], limits.totalShareNetworksUsed
)
self.assertEqual(
EXAMPLE['maxTotalShareGigabytes'], limits.maxTotalShareGigabytes
)
self.assertEqual(
EXAMPLE['maxTotalShareNetworks'], limits.maxTotalShareNetworks
)
self.assertEqual(EXAMPLE['totalSharesUsed'], limits.totalSharesUsed)
self.assertEqual(
EXAMPLE['totalShareGigabytesUsed'], limits.totalShareGigabytesUsed
)
self.assertEqual(
EXAMPLE['totalShareSnapshotsUsed'], limits.totalShareSnapshotsUsed
)
self.assertEqual(EXAMPLE['maxTotalShares'], limits.maxTotalShares)
self.assertEqual(
EXAMPLE['totalSnapshotGigabytesUsed'],
limits.totalSnapshotGigabytesUsed,
)
self.assertEqual(
EXAMPLE['maxTotalSnapshotGigabytes'],
limits.maxTotalSnapshotGigabytes,
)
self.assertEqual(
EXAMPLE['maxTotalShareSnapshots'], limits.maxTotalShareSnapshots
)
self.assertEqual(
EXAMPLE['maxTotalShareReplicas'], limits.maxTotalShareReplicas
)
self.assertEqual(
EXAMPLE['maxTotalReplicaGigabytes'],
limits.maxTotalReplicaGigabytes,
)
self.assertEqual(
EXAMPLE['totalShareReplicasUsed'], limits.totalShareReplicasUsed
)
self.assertEqual(
EXAMPLE['totalReplicaGigabytesUsed'],
limits.totalReplicaGigabytesUsed,
)

202
openstack/tests/unit/shared_file_system/v2/test_proxy.py
View File

@ -12,12 +12,10 @@
from unittest import mock
from openstack.shared_file_system.v2 import (
share_access_rule
)
from openstack.shared_file_system.v2 import _proxy
from openstack.shared_file_system.v2 import limit
from openstack.shared_file_system.v2 import share
from openstack.shared_file_system.v2 import share_access_rule
from openstack.shared_file_system.v2 import share_instance
from openstack.shared_file_system.v2 import share_network
from openstack.shared_file_system.v2 import share_network_subnet
@ -29,7 +27,6 @@ from openstack.tests.unit import test_proxy_base
class TestSharedFileSystemProxy(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestSharedFileSystemProxy, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -40,25 +37,29 @@ class TestSharedFileSystemShare(TestSharedFileSystemProxy):
self.verify_list(self.proxy.shares, share.Share)
def test_shares_detailed(self):
self.verify_list(self.proxy.shares, share.Share,
method_kwargs={"details": True, "query": 1},
expected_kwargs={"query": 1})
self.verify_list(
self.proxy.shares,
share.Share,
method_kwargs={"details": True, "query": 1},
expected_kwargs={"query": 1},
)
def test_shares_not_detailed(self):
self.verify_list(self.proxy.shares, share.Share,
method_kwargs={"details": False, "query": 1},
expected_kwargs={"query": 1})
self.verify_list(
self.proxy.shares,
share.Share,
method_kwargs={"details": False, "query": 1},
expected_kwargs={"query": 1},
)
def test_share_get(self):
self.verify_get(self.proxy.get_share, share.Share)
def test_share_delete(self):
self.verify_delete(
self.proxy.delete_share, share.Share, False)
self.verify_delete(self.proxy.delete_share, share.Share, False)
def test_share_delete_ignore(self):
self.verify_delete(
self.proxy.delete_share, share.Share, True)
self.verify_delete(self.proxy.delete_share, share.Share, True)
def test_share_create(self):
self.verify_create(self.proxy.create_share, share.Share)
@ -71,8 +72,7 @@ class TestSharedFileSystemShare(TestSharedFileSystemProxy):
self.proxy._get = mock.Mock(return_value=mock_share)
self._verify(
"openstack.shared_file_system.v2.share."
+ "Share.extend_share",
"openstack.shared_file_system.v2.share." + "Share.extend_share",
self.proxy.resize_share,
method_args=['fakeId', 20],
expected_args=[self.proxy, 20, False],
@ -83,20 +83,21 @@ class TestSharedFileSystemShare(TestSharedFileSystemProxy):
self.proxy._get = mock.Mock(return_value=mock_share)
self._verify(
"openstack.shared_file_system.v2.share."
+ "Share.shrink_share",
"openstack.shared_file_system.v2.share." + "Share.shrink_share",
self.proxy.resize_share,
method_args=['fakeId', 20],
expected_args=[self.proxy, 20],
)
def test_share_instances(self):
self.verify_list(self.proxy.share_instances,
share_instance.ShareInstance)
self.verify_list(
self.proxy.share_instances, share_instance.ShareInstance
)
def test_share_instance_get(self):
self.verify_get(self.proxy.get_share_instance,
share_instance.ShareInstance)
self.verify_get(
self.proxy.get_share_instance, share_instance.ShareInstance
)
def test_share_instance_reset(self):
self._verify(
@ -113,7 +114,8 @@ class TestSharedFileSystemShare(TestSharedFileSystemProxy):
+ "ShareInstance.force_delete",
self.proxy.delete_share_instance,
method_args=['id'],
expected_args=[self.proxy])
expected_args=[self.proxy],
)
@mock.patch("openstack.resource.wait_for_status")
def test_wait_for(self, mock_wait):
@ -122,30 +124,33 @@ class TestSharedFileSystemShare(TestSharedFileSystemProxy):
self.proxy.wait_for_status(mock_resource, 'ACTIVE')
mock_wait.assert_called_once_with(self.proxy, mock_resource,
'ACTIVE', [], 2, 120)
mock_wait.assert_called_once_with(
self.proxy, mock_resource, 'ACTIVE', [], 2, 120
)
class TestSharedFileSystemStoragePool(TestSharedFileSystemProxy):
def test_storage_pools(self):
self.verify_list(
self.proxy.storage_pools, storage_pool.StoragePool)
self.verify_list(self.proxy.storage_pools, storage_pool.StoragePool)
def test_storage_pool_detailed(self):
self.verify_list(
self.proxy.storage_pools, storage_pool.StoragePool,
self.proxy.storage_pools,
storage_pool.StoragePool,
method_kwargs={"details": True, "backend": "alpha"},
expected_kwargs={"backend": "alpha"})
expected_kwargs={"backend": "alpha"},
)
def test_storage_pool_not_detailed(self):
self.verify_list(
self.proxy.storage_pools, storage_pool.StoragePool,
self.proxy.storage_pools,
storage_pool.StoragePool,
method_kwargs={"details": False, "backend": "alpha"},
expected_kwargs={"backend": "alpha"})
expected_kwargs={"backend": "alpha"},
)
class TestUserMessageProxy(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestUserMessageProxy, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -155,70 +160,83 @@ class TestUserMessageProxy(test_proxy_base.TestProxyBase):
def test_user_messages_queried(self):
self.verify_list(
self.proxy.user_messages, user_message.UserMessage,
self.proxy.user_messages,
user_message.UserMessage,
method_kwargs={"action_id": "1"},
expected_kwargs={"action_id": "1"})
expected_kwargs={"action_id": "1"},
)
def test_user_message_get(self):
self.verify_get(self.proxy.get_user_message, user_message.UserMessage)
def test_delete_user_message(self):
self.verify_delete(
self.proxy.delete_user_message, user_message.UserMessage, False)
self.proxy.delete_user_message, user_message.UserMessage, False
)
def test_delete_user_message_true(self):
self.verify_delete(
self.proxy.delete_user_message, user_message.UserMessage, True)
self.proxy.delete_user_message, user_message.UserMessage, True
)
def test_limit(self):
self.verify_list(self.proxy.limits, limit.Limit)
class TestShareSnapshotResource(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestShareSnapshotResource, self).setUp()
self.proxy = _proxy.Proxy(self.session)
def test_share_snapshots(self):
self.verify_list(
self.proxy.share_snapshots, share_snapshot.ShareSnapshot)
self.proxy.share_snapshots, share_snapshot.ShareSnapshot
)
def test_share_snapshots_detailed(self):
self.verify_list(
self.proxy.share_snapshots,
share_snapshot.ShareSnapshot,
method_kwargs={"details": True, "name": "my_snapshot"},
expected_kwargs={"name": "my_snapshot"})
expected_kwargs={"name": "my_snapshot"},
)
def test_share_snapshots_not_detailed(self):
self.verify_list(
self.proxy.share_snapshots,
share_snapshot.ShareSnapshot,
method_kwargs={"details": False, "name": "my_snapshot"},
expected_kwargs={"name": "my_snapshot"})
expected_kwargs={"name": "my_snapshot"},
)
def test_share_snapshot_get(self):
self.verify_get(
self.proxy.get_share_snapshot, share_snapshot.ShareSnapshot)
self.proxy.get_share_snapshot, share_snapshot.ShareSnapshot
)
def test_share_snapshot_delete(self):
self.verify_delete(
self.proxy.delete_share_snapshot,
share_snapshot.ShareSnapshot, False)
share_snapshot.ShareSnapshot,
False,
)
def test_share_snapshot_delete_ignore(self):
self.verify_delete(
self.proxy.delete_share_snapshot,
share_snapshot.ShareSnapshot, True)
share_snapshot.ShareSnapshot,
True,
)
def test_share_snapshot_create(self):
self.verify_create(
self.proxy.create_share_snapshot, share_snapshot.ShareSnapshot)
self.proxy.create_share_snapshot, share_snapshot.ShareSnapshot
)
def test_share_snapshot_update(self):
self.verify_update(
self.proxy.update_share_snapshot, share_snapshot.ShareSnapshot)
self.proxy.update_share_snapshot, share_snapshot.ShareSnapshot
)
@mock.patch("openstack.resource.wait_for_delete")
def test_wait_for_delete(self, mock_wait):
@ -231,7 +249,6 @@ class TestShareSnapshotResource(test_proxy_base.TestProxyBase):
class TestShareSnapshotInstanceResource(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestShareSnapshotInstanceResource, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -239,34 +256,33 @@ class TestShareSnapshotInstanceResource(test_proxy_base.TestProxyBase):
def test_share_snapshot_instances(self):
self.verify_list(
self.proxy.share_snapshot_instances,
share_snapshot_instance.ShareSnapshotInstance)
share_snapshot_instance.ShareSnapshotInstance,
)
def test_share_snapshot_instance_detailed(self):
self.verify_list(self.proxy.share_snapshot_instances,
share_snapshot_instance.ShareSnapshotInstance,
method_kwargs={
"details": True,
"query": {'snapshot_id': 'fake'}
},
expected_kwargs={"query": {'snapshot_id': 'fake'}})
self.verify_list(
self.proxy.share_snapshot_instances,
share_snapshot_instance.ShareSnapshotInstance,
method_kwargs={"details": True, "query": {'snapshot_id': 'fake'}},
expected_kwargs={"query": {'snapshot_id': 'fake'}},
)
def test_share_snapshot_instance_not_detailed(self):
self.verify_list(self.proxy.share_snapshot_instances,
share_snapshot_instance.ShareSnapshotInstance,
method_kwargs={
"details": False,
"query": {'snapshot_id': 'fake'}
},
expected_kwargs={"query": {'snapshot_id': 'fake'}})
self.verify_list(
self.proxy.share_snapshot_instances,
share_snapshot_instance.ShareSnapshotInstance,
method_kwargs={"details": False, "query": {'snapshot_id': 'fake'}},
expected_kwargs={"query": {'snapshot_id': 'fake'}},
)
def test_share_snapshot_instance_get(self):
self.verify_get(
self.proxy.get_share_snapshot_instance,
share_snapshot_instance.ShareSnapshotInstance)
share_snapshot_instance.ShareSnapshotInstance,
)
class TestShareNetworkResource(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestShareNetworkResource, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -275,38 +291,48 @@ class TestShareNetworkResource(test_proxy_base.TestProxyBase):
self.verify_list(self.proxy.share_networks, share_network.ShareNetwork)
def test_share_networks_detailed(self):
self.verify_list(self.proxy.share_networks, share_network.ShareNetwork,
method_kwargs={"details": True, "name": "my_net"},
expected_kwargs={"name": "my_net"})
self.verify_list(
self.proxy.share_networks,
share_network.ShareNetwork,
method_kwargs={"details": True, "name": "my_net"},
expected_kwargs={"name": "my_net"},
)
def test_share_networks_not_detailed(self):
self.verify_list(self.proxy.share_networks, share_network.ShareNetwork,
method_kwargs={"details": False, "name": "my_net"},
expected_kwargs={"name": "my_net"})
self.verify_list(
self.proxy.share_networks,
share_network.ShareNetwork,
method_kwargs={"details": False, "name": "my_net"},
expected_kwargs={"name": "my_net"},
)
def test_share_network_get(self):
self.verify_get(
self.proxy.get_share_network, share_network.ShareNetwork)
self.proxy.get_share_network, share_network.ShareNetwork
)
def test_share_network_delete(self):
self.verify_delete(
self.proxy.delete_share_network, share_network.ShareNetwork, False)
self.proxy.delete_share_network, share_network.ShareNetwork, False
)
def test_share_network_delete_ignore(self):
self.verify_delete(
self.proxy.delete_share_network, share_network.ShareNetwork, True)
self.proxy.delete_share_network, share_network.ShareNetwork, True
)
def test_share_network_create(self):
self.verify_create(
self.proxy.create_share_network, share_network.ShareNetwork)
self.proxy.create_share_network, share_network.ShareNetwork
)
def test_share_network_update(self):
self.verify_update(
self.proxy.update_share_network, share_network.ShareNetwork)
self.proxy.update_share_network, share_network.ShareNetwork
)
class TestShareNetworkSubnetResource(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestShareNetworkSubnetResource, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -317,7 +343,8 @@ class TestShareNetworkSubnetResource(test_proxy_base.TestProxyBase):
share_network_subnet.ShareNetworkSubnet,
method_args=["test_share"],
expected_args=[],
expected_kwargs={"share_network_id": "test_share"})
expected_kwargs={"share_network_id": "test_share"},
)
def test_share_network_subnet_get(self):
self.verify_get(
@ -325,7 +352,8 @@ class TestShareNetworkSubnetResource(test_proxy_base.TestProxyBase):
share_network_subnet.ShareNetworkSubnet,
method_args=["fake_network_id", "fake_sub_network_id"],
expected_args=['fake_sub_network_id'],
expected_kwargs={'share_network_id': 'fake_network_id'})
expected_kwargs={'share_network_id': 'fake_network_id'},
)
def test_share_network_subnet_create(self):
self.verify_create(
@ -336,7 +364,9 @@ class TestShareNetworkSubnetResource(test_proxy_base.TestProxyBase):
expected_args=[],
expected_kwargs={
"share_network_id": "fake_network_id",
"p1": "v1"})
"p1": "v1",
},
)
def test_share_network_subnet_delete(self):
self.verify_delete(
@ -345,11 +375,11 @@ class TestShareNetworkSubnetResource(test_proxy_base.TestProxyBase):
False,
method_args=["fake_network_id", "fake_sub_network_id"],
expected_args=["fake_sub_network_id"],
expected_kwargs={'share_network_id': 'fake_network_id'})
expected_kwargs={'share_network_id': 'fake_network_id'},
)
class TestAccessRuleProxy(test_proxy_base.TestProxyBase):
def setUp(self):
super(TestAccessRuleProxy, self).setUp()
self.proxy = _proxy.Proxy(self.session)
@ -360,18 +390,21 @@ class TestAccessRuleProxy(test_proxy_base.TestProxyBase):
share_access_rule.ShareAccessRule,
method_args=["test_share"],
expected_args=[],
expected_kwargs={"share_id": "test_share"})
expected_kwargs={"share_id": "test_share"},
)
def test_access_rules_get(self):
self.verify_get(
self.proxy.get_access_rule, share_access_rule.ShareAccessRule)
self.proxy.get_access_rule, share_access_rule.ShareAccessRule
)
def test_access_rules_create(self):
self.verify_create(
self.proxy.create_access_rule,
share_access_rule.ShareAccessRule,
method_args=["share_id"],
expected_args=[])
expected_args=[],
)
def test_access_rules_delete(self):
self._verify(
@ -379,4 +412,5 @@ class TestAccessRuleProxy(test_proxy_base.TestProxyBase):
+ "ShareAccessRule.delete",
self.proxy.delete_access_rule,
method_args=['access_id', 'share_id', 'ignore_missing'],
expected_args=[self.proxy , 'share_id'])
expected_args=[self.proxy, 'share_id'],
)

98
openstack/tests/unit/shared_file_system/v2/test_share.py
View File

@ -49,12 +49,11 @@ EXAMPLE = {
"is_mounting_snapshot_supported": True,
"progress": "100%",
"share_server_id": None,
"host": "new@denver#lvm-single-pool"
"host": "new@denver#lvm-single-pool",
}
class TestShares(base.TestCase):
def test_basic(self):
shares_resource = share.Share()
self.assertEqual('shares', shares_resource.resources_key)
@ -69,50 +68,65 @@ class TestShares(base.TestCase):
shares_resource = share.Share(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], shares_resource.id)
self.assertEqual(EXAMPLE['size'], shares_resource.size)
self.assertEqual(EXAMPLE['availability_zone'],
shares_resource.availability_zone)
self.assertEqual(
EXAMPLE['availability_zone'], shares_resource.availability_zone
)
self.assertEqual(EXAMPLE['created_at'], shares_resource.created_at)
self.assertEqual(EXAMPLE['status'], shares_resource.status)
self.assertEqual(EXAMPLE['name'], shares_resource.name)
self.assertEqual(EXAMPLE['description'],
shares_resource.description)
self.assertEqual(EXAMPLE['description'], shares_resource.description)
self.assertEqual(EXAMPLE['project_id'], shares_resource.project_id)
self.assertEqual(EXAMPLE['snapshot_id'], shares_resource.snapshot_id)
self.assertEqual(EXAMPLE['share_network_id'],
shares_resource.share_network_id)
self.assertEqual(EXAMPLE['share_protocol'],
shares_resource.share_protocol)
self.assertEqual(
EXAMPLE['share_network_id'], shares_resource.share_network_id
)
self.assertEqual(
EXAMPLE['share_protocol'], shares_resource.share_protocol
)
self.assertEqual(EXAMPLE['metadata'], shares_resource.metadata)
self.assertEqual(EXAMPLE['share_type'], shares_resource.share_type)
self.assertEqual(EXAMPLE['is_public'], shares_resource.is_public)
self.assertEqual(EXAMPLE['is_snapshot_supported'],
shares_resource.is_snapshot_supported)
self.assertEqual(
EXAMPLE['is_snapshot_supported'],
shares_resource.is_snapshot_supported,
)
self.assertEqual(EXAMPLE['task_state'], shares_resource.task_state)
self.assertEqual(EXAMPLE['share_type_name'],
shares_resource.share_type_name)
self.assertEqual(EXAMPLE['access_rules_status'],
shares_resource.access_rules_status)
self.assertEqual(EXAMPLE['replication_type'],
shares_resource.replication_type)
self.assertEqual(EXAMPLE['is_replicated'],
shares_resource.is_replicated)
self.assertEqual(
EXAMPLE['share_type_name'], shares_resource.share_type_name
)
self.assertEqual(
EXAMPLE['access_rules_status'], shares_resource.access_rules_status
)
self.assertEqual(
EXAMPLE['replication_type'], shares_resource.replication_type
)
self.assertEqual(
EXAMPLE['is_replicated'], shares_resource.is_replicated
)
self.assertEqual(EXAMPLE['user_id'], shares_resource.user_id)
self.assertEqual(EXAMPLE[
'is_creating_new_share_from_snapshot_supported'],
(shares_resource.is_creating_new_share_from_snapshot_supported))
self.assertEqual(EXAMPLE['is_reverting_to_snapshot_supported'],
shares_resource.is_reverting_to_snapshot_supported)
self.assertEqual(EXAMPLE['share_group_id'],
shares_resource.share_group_id)
self.assertEqual(EXAMPLE[
'source_share_group_snapshot_member_id'],
shares_resource.source_share_group_snapshot_member_id)
self.assertEqual(EXAMPLE['is_mounting_snapshot_supported'],
shares_resource.is_mounting_snapshot_supported)
self.assertEqual(EXAMPLE['progress'],
shares_resource.progress)
self.assertEqual(EXAMPLE['share_server_id'],
shares_resource.share_server_id)
self.assertEqual(
EXAMPLE['is_creating_new_share_from_snapshot_supported'],
(shares_resource.is_creating_new_share_from_snapshot_supported),
)
self.assertEqual(
EXAMPLE['is_reverting_to_snapshot_supported'],
shares_resource.is_reverting_to_snapshot_supported,
)
self.assertEqual(
EXAMPLE['share_group_id'], shares_resource.share_group_id
)
self.assertEqual(
EXAMPLE['source_share_group_snapshot_member_id'],
shares_resource.source_share_group_snapshot_member_id,
)
self.assertEqual(
EXAMPLE['is_mounting_snapshot_supported'],
shares_resource.is_mounting_snapshot_supported,
)
self.assertEqual(EXAMPLE['progress'], shares_resource.progress)
self.assertEqual(
EXAMPLE['share_server_id'], shares_resource.share_server_id
)
self.assertEqual(EXAMPLE['host'], shares_resource.host)
@ -139,8 +153,8 @@ class TestShareActions(TestShares):
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, json=body, headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)
def test_extend_share(self):
sot = share.Share(**EXAMPLE)
@ -153,8 +167,8 @@ class TestShareActions(TestShares):
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, json=body, headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)
def test_revert_to_snapshot(self):
sot = share.Share(**EXAMPLE)
@ -167,5 +181,5 @@ class TestShareActions(TestShares):
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, json=body, headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)

15
openstack/tests/unit/shared_file_system/v2/test_share_access_rule.py
View File

@ -23,26 +23,21 @@ EXAMPLE = {
"access_key": None,
"created_at": "2021-09-12T02:01:04.000000",
"updated_at": "2021-09-12T02:01:04.000000",
"metadata": {
"key1": "value1",
"key2": "value2"}
"metadata": {"key1": "value1", "key2": "value2"},
}
class TestShareAccessRule(base.TestCase):
def test_basic(self):
rules_resource = share_access_rule.ShareAccessRule()
self.assertEqual('access_list', rules_resource.resources_key)
self.assertEqual('/share-access-rules', rules_resource.base_path)
self.assertTrue(rules_resource.allow_list)
self.assertDictEqual({
"limit": "limit",
"marker": "marker",
"share_id": "share_id"
},
rules_resource._query_mapping._mapping)
self.assertDictEqual(
{"limit": "limit", "marker": "marker", "share_id": "share_id"},
rules_resource._query_mapping._mapping,
)
def test_make_share_access_rules(self):
rules_resource = share_access_rule.ShareAccessRule(**EXAMPLE)

15
openstack/tests/unit/shared_file_system/v2/test_share_export_locations.py
View File

@ -17,21 +17,23 @@ from openstack.tests.unit import base
IDENTIFIER = '08a87d37-5ca2-4308-86c5-cba06d8d796c'
EXAMPLE = {
"id": "f87589cb-f4bc-4a9b-b481-ab701206eb85",
"path": ("199.19.213.225:/opt/stack/data/manila/mnt/"
"share-6ba490c5-5225-4c3b-9982-14b8f475c6d9"),
"path": (
"199.19.213.225:/opt/stack/data/manila/mnt/"
"share-6ba490c5-5225-4c3b-9982-14b8f475c6d9"
),
"preferred": False,
"share_instance_id": "6ba490c5-5225-4c3b-9982-14b8f475c6d9",
"is_admin_only": False
"is_admin_only": False,
}
class TestShareExportLocations(base.TestCase):
def test_basic(self):
export = el.ShareExportLocation()
self.assertEqual('export_locations', export.resources_key)
self.assertEqual(
'/shares/%(share_id)s/export_locations', export.base_path)
'/shares/%(share_id)s/export_locations', export.base_path
)
self.assertTrue(export.allow_list)
def test_share_export_locations(self):
@ -40,5 +42,6 @@ class TestShareExportLocations(base.TestCase):
self.assertEqual(EXAMPLE['path'], export.path)
self.assertEqual(EXAMPLE['preferred'], export.is_preferred)
self.assertEqual(
EXAMPLE['share_instance_id'], export.share_instance_id)
EXAMPLE['share_instance_id'], export.share_instance_id
)
self.assertEqual(EXAMPLE['is_admin_only'], export.is_admin)

56
openstack/tests/unit/shared_file_system/v2/test_share_instance.py
View File

@ -31,16 +31,16 @@ EXAMPLE = {
"share_server_id": "ba11930a-bf1a-4aa7-bae4-a8dfbaa3cc73",
"host": "manila2@generic1#GENERIC1",
"access_rules_status": "active",
"id": IDENTIFIER
"id": IDENTIFIER,
}
class TestShareInstances(base.TestCase):
def test_basic(self):
share_instance_resource = share_instance.ShareInstance()
self.assertEqual('share_instances',
share_instance_resource.resources_key)
self.assertEqual(
'share_instances', share_instance_resource.resources_key
)
self.assertEqual('/share_instances', share_instance_resource.base_path)
self.assertTrue(share_instance_resource.allow_list)
self.assertFalse(share_instance_resource.allow_create)
@ -53,26 +53,36 @@ class TestShareInstances(base.TestCase):
self.assertEqual(EXAMPLE['status'], share_instance_resource.status)
self.assertEqual(EXAMPLE['progress'], share_instance_resource.progress)
self.assertEqual(EXAMPLE['share_id'], share_instance_resource.share_id)
self.assertEqual(EXAMPLE['availability_zone'],
share_instance_resource.availability_zone)
self.assertEqual(EXAMPLE['replica_state'],
share_instance_resource.replica_state)
self.assertEqual(EXAMPLE['created_at'],
share_instance_resource.created_at)
self.assertEqual(EXAMPLE['cast_rules_to_readonly'],
share_instance_resource.cast_rules_to_readonly)
self.assertEqual(EXAMPLE['share_network_id'],
share_instance_resource.share_network_id)
self.assertEqual(EXAMPLE['share_server_id'],
share_instance_resource.share_server_id)
self.assertEqual(
EXAMPLE['availability_zone'],
share_instance_resource.availability_zone,
)
self.assertEqual(
EXAMPLE['replica_state'], share_instance_resource.replica_state
)
self.assertEqual(
EXAMPLE['created_at'], share_instance_resource.created_at
)
self.assertEqual(
EXAMPLE['cast_rules_to_readonly'],
share_instance_resource.cast_rules_to_readonly,
)
self.assertEqual(
EXAMPLE['share_network_id'],
share_instance_resource.share_network_id,
)
self.assertEqual(
EXAMPLE['share_server_id'], share_instance_resource.share_server_id
)
self.assertEqual(EXAMPLE['host'], share_instance_resource.host)
self.assertEqual(EXAMPLE['access_rules_status'],
share_instance_resource.access_rules_status)
self.assertEqual(
EXAMPLE['access_rules_status'],
share_instance_resource.access_rules_status,
)
self.assertEqual(EXAMPLE['id'], share_instance_resource.id)
class TestShareInstanceActions(TestShareInstances):
def setUp(self):
super(TestShareInstanceActions, self).setUp()
self.resp = mock.Mock()
@ -94,8 +104,8 @@ class TestShareInstanceActions(TestShareInstances):
body = {"reset_status": {"status": 'active'}}
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, json=body, headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)
def test_force_delete(self):
sot = share_instance.ShareInstance(**EXAMPLE)
@ -107,5 +117,5 @@ class TestShareInstanceActions(TestShareInstances):
body = {'force_delete': None}
headers = {'Accept': ''}
self.sess.post.assert_called_with(
url, json=body, headers=headers,
microversion=microversion)
url, json=body, headers=headers, microversion=microversion
)

39
openstack/tests/unit/shared_file_system/v2/test_share_network.py
View File

@ -21,12 +21,11 @@ EXAMPLE = {
"description": "My share network",
"created_at": "2021-06-10T10:11:17.291981",
"updated_at": None,
"share_network_subnets": []
"share_network_subnets": [],
}
class TestShareNetwork(base.TestCase):
def test_basic(self):
networks = share_network.ShareNetwork()
self.assertEqual('share_networks', networks.resources_key)
@ -38,28 +37,28 @@ class TestShareNetwork(base.TestCase):
self.assertTrue(networks.allow_delete)
self.assertFalse(networks.allow_head)
self.assertDictEqual({
"limit": "limit",
"marker": "marker",
"project_id": "project_id",
"created_since": "created_since",
"created_before": "created_before",
"offset": "offset",
"security_service_id": "security_service_id",
"project_id": "project_id",
"all_projects": "all_tenants",
"name": "name",
"description": "description"
},
networks._query_mapping._mapping)
self.assertDictEqual(
{
"limit": "limit",
"marker": "marker",
"project_id": "project_id",
"created_since": "created_since",
"created_before": "created_before",
"offset": "offset",
"security_service_id": "security_service_id",
"project_id": "project_id",
"all_projects": "all_tenants",
"name": "name",
"description": "description",
},
networks._query_mapping._mapping,
)
def test_share_network(self):
networks = share_network.ShareNetwork(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], networks.id)
self.assertEqual(EXAMPLE['name'], networks.name)
self.assertEqual(EXAMPLE['project_id'], networks.project_id)
self.assertEqual(
EXAMPLE['description'], networks.description)
self.assertEqual(
EXAMPLE['created_at'], networks.created_at)
self.assertEqual(EXAMPLE['description'], networks.description)
self.assertEqual(EXAMPLE['created_at'], networks.created_at)
self.assertEqual(EXAMPLE['updated_at'], networks.updated_at)

39
openstack/tests/unit/shared_file_system/v2/test_share_network_subnet.py
View File

@ -28,36 +28,43 @@ EXAMPLE = {
"cidr": None,
"network_type": None,
"mtu": None,
"gateway": None
"gateway": None,
}
class TestShareNetworkSubnet(base.TestCase):
def test_basic(self):
SNS_resource = SNS.ShareNetworkSubnet()
self.assertEqual('share_network_subnets', SNS_resource.resources_key)
self.assertEqual('/share-networks/%(share_network_id)s/subnets',
SNS_resource.base_path)
self.assertEqual(
'/share-networks/%(share_network_id)s/subnets',
SNS_resource.base_path,
)
self.assertTrue(SNS_resource.allow_list)
def test_make_share_network_subnet(self):
SNS_resource = SNS.ShareNetworkSubnet(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], SNS_resource.id)
self.assertEqual(EXAMPLE['availability_zone'],
SNS_resource.availability_zone)
self.assertEqual(EXAMPLE['share_network_id'],
SNS_resource.share_network_id)
self.assertEqual(EXAMPLE['share_network_name'],
SNS_resource.share_network_name)
self.assertEqual(
EXAMPLE['availability_zone'], SNS_resource.availability_zone
)
self.assertEqual(
EXAMPLE['share_network_id'], SNS_resource.share_network_id
)
self.assertEqual(
EXAMPLE['share_network_name'], SNS_resource.share_network_name
)
self.assertEqual(EXAMPLE['created_at'], SNS_resource.created_at)
self.assertEqual(EXAMPLE['segmentation_id'],
SNS_resource.segmentation_id)
self.assertEqual(EXAMPLE['neutron_subnet_id'],
SNS_resource.neutron_subnet_id)
self.assertEqual(
EXAMPLE['segmentation_id'], SNS_resource.segmentation_id
)
self.assertEqual(
EXAMPLE['neutron_subnet_id'], SNS_resource.neutron_subnet_id
)
self.assertEqual(EXAMPLE['updated_at'], SNS_resource.updated_at)
self.assertEqual(EXAMPLE['neutron_net_id'],
SNS_resource.neutron_net_id)
self.assertEqual(
EXAMPLE['neutron_net_id'], SNS_resource.neutron_net_id
)
self.assertEqual(EXAMPLE['ip_version'], SNS_resource.ip_version)
self.assertEqual(EXAMPLE['cidr'], SNS_resource.cidr)
self.assertEqual(EXAMPLE['network_type'], SNS_resource.network_type)

35
openstack/tests/unit/shared_file_system/v2/test_share_snapshot.py
View File

@ -25,41 +25,36 @@ EXAMPLE = {
"share_size": 1,
"id": "6d221c1d-0200-461e-8d20-24b4776b9ddb",
"project_id": "cadd7139bc3148b8973df097c0911016",
"size": 1
"size": 1,
}
class TestShareSnapshot(base.TestCase):
def test_basic(self):
snapshot_resource = share_snapshot.ShareSnapshot()
self.assertEqual(
'snapshots', snapshot_resource.resources_key)
self.assertEqual('snapshots', snapshot_resource.resources_key)
self.assertEqual('/snapshots', snapshot_resource.base_path)
self.assertTrue(snapshot_resource.allow_list)
self.assertDictEqual({
"limit": "limit",
"marker": "marker",
"snapshot_id": "snapshot_id"
},
snapshot_resource._query_mapping._mapping)
self.assertDictEqual(
{
"limit": "limit",
"marker": "marker",
"snapshot_id": "snapshot_id",
},
snapshot_resource._query_mapping._mapping,
)
def test_make_share_snapshot(self):
snapshot_resource = share_snapshot.ShareSnapshot(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], snapshot_resource.id)
self.assertEqual(EXAMPLE['share_id'], snapshot_resource.share_id)
self.assertEqual(EXAMPLE['user_id'],
snapshot_resource.user_id)
self.assertEqual(EXAMPLE['user_id'], snapshot_resource.user_id)
self.assertEqual(EXAMPLE['created_at'], snapshot_resource.created_at)
self.assertEqual(EXAMPLE['status'], snapshot_resource.status)
self.assertEqual(EXAMPLE['name'], snapshot_resource.name)
self.assertEqual(
EXAMPLE['description'], snapshot_resource.description)
self.assertEqual(
EXAMPLE['share_proto'], snapshot_resource.share_proto)
self.assertEqual(
EXAMPLE['share_size'], snapshot_resource.share_size)
self.assertEqual(
EXAMPLE['project_id'], snapshot_resource.project_id)
self.assertEqual(EXAMPLE['description'], snapshot_resource.description)
self.assertEqual(EXAMPLE['share_proto'], snapshot_resource.share_proto)
self.assertEqual(EXAMPLE['share_size'], snapshot_resource.share_size)
self.assertEqual(EXAMPLE['project_id'], snapshot_resource.project_id)
self.assertEqual(EXAMPLE['size'], snapshot_resource.size)

9
openstack/tests/unit/shared_file_system/v2/test_share_snapshot_instance.py
View File

@ -22,12 +22,11 @@ EXAMPLE = {
"created_at": "2021-06-04T00:44:52.000000",
"id": "275516e8-c998-4e78-a41e-7dd3a03e71cd",
"provider_location": "/path/to/fake...",
"updated_at": "2017-06-04T00:44:54.000000"
"updated_at": "2017-06-04T00:44:54.000000",
}
class TestShareSnapshotInstances(base.TestCase):
def test_basic(self):
instances = share_snapshot_instance.ShareSnapshotInstance()
self.assertEqual('snapshot_instance', instances.resource_key)
@ -40,11 +39,13 @@ class TestShareSnapshotInstances(base.TestCase):
self.assertEqual(EXAMPLE['id'], instance.id)
self.assertEqual(EXAMPLE['share_id'], instance.share_id)
self.assertEqual(
EXAMPLE['share_instance_id'], instance.share_instance_id)
EXAMPLE['share_instance_id'], instance.share_instance_id
)
self.assertEqual(EXAMPLE['snapshot_id'], instance.snapshot_id)
self.assertEqual(EXAMPLE['status'], instance.status)
self.assertEqual(EXAMPLE['progress'], instance.progress)
self.assertEqual(EXAMPLE['created_at'], instance.created_at)
self.assertEqual(EXAMPLE['updated_at'], instance.updated_at)
self.assertEqual(
EXAMPLE['provider_location'], instance.provider_location)
EXAMPLE['provider_location'], instance.provider_location
)

32
openstack/tests/unit/shared_file_system/v2/test_storage_pool.py
View File

@ -40,29 +40,30 @@ EXAMPLE = {
"replication_domain": None,
"sg_consistent_snapshot_support": "pool",
"ipv4_support": True,
"ipv6_support": False}
"ipv6_support": False,
},
}
class TestStoragePool(base.TestCase):
def test_basic(self):
pool_resource = storage_pool.StoragePool()
self.assertEqual('pools', pool_resource.resources_key)
self.assertEqual(
'/scheduler-stats/pools', pool_resource.base_path)
self.assertEqual('/scheduler-stats/pools', pool_resource.base_path)
self.assertTrue(pool_resource.allow_list)
self.assertDictEqual({
'pool': 'pool',
'backend': 'backend',
'host': 'host',
'limit': 'limit',
'marker': 'marker',
'capabilities': 'capabilities',
'share_type': 'share_type',
},
pool_resource._query_mapping._mapping)
self.assertDictEqual(
{
'pool': 'pool',
'backend': 'backend',
'host': 'host',
'limit': 'limit',
'marker': 'marker',
'capabilities': 'capabilities',
'share_type': 'share_type',
},
pool_resource._query_mapping._mapping,
)
def test_make_storage_pool(self):
pool_resource = storage_pool.StoragePool(**EXAMPLE)
@ -70,5 +71,4 @@ class TestStoragePool(base.TestCase):
self.assertEqual(EXAMPLE['host'], pool_resource.host)
self.assertEqual(EXAMPLE['name'], pool_resource.name)
self.assertEqual(EXAMPLE['backend'], pool_resource.backend)
self.assertEqual(
EXAMPLE['capabilities'], pool_resource.capabilities)
self.assertEqual(EXAMPLE['capabilities'], pool_resource.capabilities)

32
openstack/tests/unit/shared_file_system/v2/test_user_message.py
View File

@ -29,12 +29,12 @@ EXAMPLE = {
"user_message": (
"allocate host: No storage could be allocated"
"for this share request, Capabilities filter"
"didn't succeed.")
"didn't succeed."
),
}
class TestUserMessage(base.TestCase):
def test_basic(self):
message = user_message.UserMessage()
self.assertEqual('messages', message.resources_key)
@ -46,29 +46,21 @@ class TestUserMessage(base.TestCase):
self.assertTrue(message.allow_fetch)
self.assertFalse(message.allow_head)
self.assertDictEqual({
"limit": "limit",
"marker": "marker",
"message_id": "message_id"
},
message._query_mapping._mapping)
self.assertDictEqual(
{"limit": "limit", "marker": "marker", "message_id": "message_id"},
message._query_mapping._mapping,
)
def test_user_message(self):
messages = user_message.UserMessage(**EXAMPLE)
self.assertEqual(EXAMPLE['id'], messages.id)
self.assertEqual(EXAMPLE['resource_id'], messages.resource_id)
self.assertEqual(
EXAMPLE['message_level'], messages.message_level)
self.assertEqual(
EXAMPLE['user_message'], messages.user_message)
self.assertEqual(EXAMPLE['message_level'], messages.message_level)
self.assertEqual(EXAMPLE['user_message'], messages.user_message)
self.assertEqual(EXAMPLE['expires_at'], messages.expires_at)
self.assertEqual(EXAMPLE['detail_id'], messages.detail_id)
self.assertEqual(EXAMPLE['created_at'], messages.created_at)
self.assertEqual(
EXAMPLE['request_id'], messages.request_id)
self.assertEqual(
EXAMPLE['project_id'], messages.project_id)
self.assertEqual(
EXAMPLE['resource_type'], messages.resource_type)
self.assertEqual(
EXAMPLE['action_id'], messages.action_id)
self.assertEqual(EXAMPLE['request_id'], messages.request_id)
self.assertEqual(EXAMPLE['project_id'], messages.project_id)
self.assertEqual(EXAMPLE['resource_type'], messages.resource_type)
self.assertEqual(EXAMPLE['action_id'], messages.action_id)