openstack/openstacksdk
Code Issues Proposed changes

Extend has_version function to accept version parameter

Browse Source 
We might need not only to check whether the service is supported, but
also whether specific version of the service is working.
Extend functional tests of the block-storage to actually check specific
version.

Change-Id: Ie71c4f818d5adc82466e06d994c258af271b5793
This commit is contained in:
Artem Goncharov 2021-07-22 18:00:25 +02:00
parent 9dc4a84114
commit 0bc47cdb93
3 changed files with 13 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
openstack/cloud/openstackcloud.py
View File

@ -712,9 +712,11 @@ class _OpenStackCloudMixin:
def get_name(self):
return self.name
def get_session_endpoint(self, service_key):
def get_session_endpoint(self, service_key, **kwargs):
if not kwargs:
kwargs = {}
try:
return self.config.get_session_endpoint(service_key)
return self.config.get_session_endpoint(service_key, **kwargs)
except keystoneauth1.exceptions.catalog.EndpointNotFound as e:
self.log.debug(
"Endpoint not found in %s cloud: %s", self.name, str(e))
@ -731,7 +733,7 @@ class _OpenStackCloudMixin:
error=str(e)))
return endpoint
def has_service(self, service_key):
def has_service(self, service_key, version=None):
if not self.config.has_service(service_key):
# TODO(mordred) add a stamp here so that we only report this once
if not (service_key in self._disable_warnings
@ -742,7 +744,12 @@ class _OpenStackCloudMixin:
self._disable_warnings[service_key] = True
return False
try:
endpoint = self.get_session_endpoint(service_key)
kwargs = dict()
# If a specific version was requested - try it
if version is not None:
kwargs['min_version'] = version
kwargs['max_version'] = version
endpoint = self.get_session_endpoint(service_key, **kwargs)
except exc.OpenStackCloudException:
return False
if endpoint:

2
openstack/tests/functional/block_storage/v2/base.py
View File

@ -22,5 +22,5 @@ class BaseBlockStorageTest(base.BaseFunctionalTest):
self._set_user_cloud(block_storage_api_version='2')
self._set_operator_cloud(block_storage_api_version='2')
if not self.user_cloud.has_service('block-storage'):
if not self.user_cloud.has_service('block-storage', '2'):
self.skipTest('block-storage service not supported by cloud')

2
openstack/tests/functional/block_storage/v3/base.py
View File

@ -22,5 +22,5 @@ class BaseBlockStorageTest(base.BaseFunctionalTest):
self._set_user_cloud(block_storage_api_version='3')
self._set_operator_cloud(block_storage_api_version='3')
if not self.user_cloud.has_service('block-storage'):
if not self.user_cloud.has_service('block-storage', '3'):
self.skipTest('block-storage service not supported by cloud')