Automatic Tempest Configuration Generator
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
python-tempestconf/config_tempest/services/base.py

172 lines
5.5 KiB

# Copyright 2013 Red Hat, Inc.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
import re
import urllib3
from six.moves import urllib
from config_tempest.constants import LOG
MULTIPLE_SLASH = re.compile(r'/+')
class ServiceError(Exception):
pass
class Service(object):
def __init__(self, name, service_url, token, disable_ssl_validation,
client=None):
self.name = name
self.service_url = service_url
self.headers = {'Accept': 'application/json', 'X-Auth-Token': token}
self.disable_ssl_validation = disable_ssl_validation
self.client = client
self.extensions = []
self.versions = []
def do_get(self, url, top_level=False, top_level_path=""):
parts = list(urllib.parse.urlparse(url))
# 2 is the path offset
if top_level:
parts[2] = '/' + top_level_path
parts[2] = MULTIPLE_SLASH.sub('/', parts[2])
url = urllib.parse.urlunparse(parts)
try:
if self.disable_ssl_validation:
urllib3.disable_warnings()
http = urllib3.PoolManager(cert_reqs='CERT_NONE')
else:
http = urllib3.PoolManager()
r = http.request('GET', url, headers=self.headers)
except Exception as e:
LOG.error("Request on service '%s' with url '%s' failed",
(self.name, url))
raise e
if r.status >= 400:
raise ServiceError("Request on service '%s' with url '%s' failed"
" with code %d" % (self.name, url, r.status))
return r.data
def set_extensions(self):
self.extensions = []
def set_versions(self):
self.versions = []
def set_availability(self, conf, available):
"""Sets service's availability.
The services's codename will be set to desired value under
[service_available] section in tempest.conf during the services
discovery process.
"""
try:
conf.set('service_available', self.get_codename(), str(available))
except NotImplementedError:
pass
def get_extensions(self):
return self.extensions
@staticmethod
def get_service_name():
"""Return the service name.
This returns a list because you can have different services for the
same type, like volume, volumev2, volumev3
"""
return []
def get_versions(self):
"""Return the versions available for each service.
This doesn't mean tempestconf supports all these versions. Only that
the service has these api versions enabled.
"""
return self.versions
def set_default_tempest_options(self, conf):
pass
def get_supported_versions(self):
"""Return the versions supported by tempestconf.
The server might have older or newer versions that could not be
supported by tempestconf.
"""
return []
@staticmethod
def get_codename():
"""Return the service_available name of the service.
This name is used when setting service availability in
set_availability method. If the method is not implemented, service
availability is not set.
"""
raise NotImplementedError
def get_feature_name(self):
"""Return the name of service used in <service>-feature-enabled.
Some services have the -feature-enabled option in tempest, that
diverges from the service name. The main example is object-store
service where the <service>-feature-enabled is object-storage.
"""
return self.name
def get_service_extension_key(self):
"""Return the extension key for a particular service"""
return None
def get_unversioned_service_name(self):
"""Return name of service without versions.
Some services are versioned like volumev2 and volumev3, we try to
discover these services checking the supported versions, so we need
to know the unversioned service name for this.
The default value is the name of the service.
"""
return self.name
class VersionedService(Service):
def set_versions(self, top_level=True):
body = self.do_get(self.service_url, top_level=top_level)
body = json.loads(body)
self.versions = self.deserialize_versions(body)
def deserialize_versions(self, body):
versions = []
for version in body['versions']:
if version['status'] != "DEPRECATED":
versions.append(version)
return list(map(lambda x: x['id'], versions))
def no_port_cut_url(self):
# if there is no port defined, cut the url from version to the end
u = urllib3.util.parse_url(self.service_url)
url = self.service_url
if u.port is None:
found = re.findall(r'v\d', url)
if len(found) > 0:
index = url.index(found[0])
url = self.service_url[:index]
return (url, u.port is not None)