OpenStack Compute (Nova)
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.

549 lines
22 KiB

# Copyright 2012 Nebula, Inc.
# Copyright 2013 IBM Corp.
# 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
# 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 os
import pprint
import re
from oslo_serialization import jsonutils
from nova import test
from nova.tests.functional import integrated_helpers
PROJECT_ID = "6f70656e737461636b20342065766572"
# for pretty printing errors
pp = pprint.PrettyPrinter(indent=4)
class NoMatch(test.TestingException):
def pretty_data(data):
data = jsonutils.dumps(jsonutils.loads(data), sort_keys=True,
return '\n'.join(line.rstrip() for line in data.split('\n')).strip()
def objectify(data):
if not data:
return {}
# NOTE(sdague): templates will contain values like %(foo)s
# throughout them. If these are inside of double quoted
# strings, life is good, and we can treat it just like valid
# json to load it to python.
# However we've got some fields which are ints, like
# aggregate_id. This means we've got a snippet in the sample
# that looks like:
# "id": %(aggregate_id)s,
# which is not valid json, and will explode. We do a quick and
# dirty transform of this to:
# "id": "%(int:aggregate_id)s",
# That makes it valid data to convert to json, but keeps
# around the information that we need to drop those strings
# later. The regex anchors from the ': ', as all of these will
# be top rooted keys.
data = re.sub(r'(\: )%\((.+)\)s([^"])', r'\1"%(int:\2)s"\3', data)
return jsonutils.loads(data)
class ApiSampleTestBase(integrated_helpers._IntegratedTestBase):
sample_dir = None
microversion = None
_use_common_server_api_samples = False
def __init__(self, *args, **kwargs):
super(ApiSampleTestBase, self).__init__(*args, **kwargs)
self.subs = {} # TODO(auggy): subs should really be a class
def subs(self):
return self._subs
def subs(self, value):
non_strings = \
{k: v for k, v in value.items() if
(not k == 'compute_host') and
(not isinstance(v, str))}
if len(non_strings) > 0:
raise TypeError("subs can't contain non-string values:"
"\n%(non_strings)s" %
{'non_strings': non_strings})
self._subs = value
def _get_sample_path(cls, name, dirname, suffix='', api_version=None):
parts = [dirname]
# Note(gmann): if _use_common_server_api_samples is set to True
# then common server sample files present in 'servers' directory
# will be used. As of now it is being used for server POST request
# to avoid duplicate copy of server req and resp sample files.
# Example - ServersSampleBase's _post_server method.
if cls._use_common_server_api_samples:
if api_version:
parts.append('v' + api_version)
parts.append(name + ".json" + suffix)
return os.path.join(*parts)
def _get_sample(cls, name, api_version=None):
dirname = os.path.dirname(os.path.abspath(__file__))
dirname = os.path.normpath(os.path.join(dirname,
return cls._get_sample_path(name, dirname, api_version=api_version)
def _get_template(cls, name, api_version=None):
dirname = os.path.dirname(os.path.abspath(__file__))
dirname = os.path.normpath(os.path.join(dirname,
return cls._get_sample_path(name, dirname, suffix='.tpl',
def _read_template(self, name):
template = self._get_template(name, self.microversion)
with open(template) as inf:
def _write_template(self, name, data):
with open(self._get_template(name,
self.microversion), 'w') as outf:
def _write_sample(self, name, data):
sample_file = self._get_sample(name, self.microversion)
os.makedirs(os.path.dirname(sample_file), exist_ok = True)
with open(sample_file, 'w') as outf:
def _compare_result(self, expected, result, result_str):
matched_value = None
# None
if expected is None:
if result is None:
elif result == u'':
pass # TODO(auggy): known issue Bug#1544720
raise NoMatch('%(result_str)s: Expected None, got %(result)s.'
% {'result_str': result_str, 'result': result})
# dictionary
elif isinstance(expected, dict):
if not isinstance(result, dict):
raise NoMatch('%(result_str)s: %(result)s is not a dict.'
% {'result_str': result_str, 'result': result})
ex_keys = sorted(expected.keys())
res_keys = sorted(result.keys())
if ex_keys != res_keys:
ex_delta = []
res_delta = []
for key in ex_keys:
if key not in res_keys:
for key in res_keys:
if key not in ex_keys:
raise NoMatch(
'Dictionary key mismatch:\n'
'Extra key(s) in template:\n%(ex_delta)s\n'
'Extra key(s) in %(result_str)s:\n%(res_delta)s\n' %
{'ex_delta': ex_delta, 'result_str': result_str,
'res_delta': res_delta})
for key in ex_keys:
# TODO(auggy): pass key name along as well for error reporting
res = self._compare_result(expected[key], result[key],
matched_value = res or matched_value
# list
elif isinstance(expected, list):
if not isinstance(result, list):
raise NoMatch(
'%(result_str)s: %(result)s is not a list.' %
{'result_str': result_str, 'result': result})
expected = expected[:]
extra = []
# if it's a list of 1, do the simple compare which gives a
# better error message.
if len(result) == len(expected) == 1:
return self._compare_result(expected[0], result[0], result_str)
# This is clever enough to need some explanation. What we
# are doing here is looping the result list, and trying to
# compare it to every item in the expected list. If there
# is more than one, we're going to get fails. We ignore
# those. But every time we match an expected we drop it,
# and break to the next iteration. Every time we hit the
# end of the iteration, we add our results into a bucket
# of non matched.
# This results in poor error messages because we don't
# really know why the elements failed to match each
# other. A more complicated diff might be nice.
for res_obj in result:
for i, ex_obj in enumerate(expected):
matched_value = self._compare_result(ex_obj,
del expected[i]
except NoMatch:
error = []
if expected:
error.append('Extra list items in template:')
error.extend([repr(o) for o in expected])
if extra:
error.append('Extra list items in %(result_str)s:' %
{'result_str': result_str})
error.extend([repr(o) for o in extra])
if error:
raise NoMatch('\n'.join(error))
# template string
elif isinstance(expected, str) and '%' in expected:
# NOTE(vish): escape stuff for regex
for char in '[]<>?':
expected = expected.replace(char, '\\%s' % char)
# NOTE(vish): special handling of subs that are not quoted. We are
# expecting an int but we had to pass in a string
# so the json would parse properly.
if expected.startswith("%(int:"):
result = str(result)
expected = expected.replace('int:', '')
expected = expected % self.subs
expected = '^%s$' % expected
match = re.match(expected, result)
except TypeError as e:
raise NoMatch(
'Values do not match:\n'
'Template: %(expected)s\n%(result_str)s: %(result)s\n'
'Error: %(error)s' %
{'expected': expected, 'result_str': result_str,
'result': result, 'error': e})
if not match:
raise NoMatch(
'Values do not match:\n'
'Template: %(expected)s\n%(result_str)s: %(result)s' %
{'expected': expected, 'result_str': result_str,
'result': result})
matched_value ='id')
except IndexError:
if match.groups():
matched_value = match.groups()[0]
# string
elif isinstance(expected, str):
# NOTE(danms): Ignore whitespace in this comparison
expected = expected.strip()
if isinstance(result, str):
result = result.strip()
if expected != result:
# NOTE(tdurakov):this attempt to parse string as JSON
# is needed for correct comparison of hypervisor.cpu_info,
# which is stringified JSON object
# TODO(tdurakov): remove this check as soon as
# hypervisor.cpu_info become common JSON object in REST API.
expected = objectify(expected)
result = objectify(result)
return self._compare_result(expected, result,
except ValueError:
raise NoMatch(
'Values do not match:\n'
'Template: %(expected)s\n%(result_str)s: '
'%(result)s' % {'expected': expected,
'result_str': result_str,
'result': result})
# int
elif isinstance(expected, (int, float)):
if expected != result:
raise NoMatch(
'Values do not match:\n'
'Template: %(expected)s\n%(result_str)s: '
'%(result)s' % {'expected': expected,
'result_str': result_str,
'result': result})
raise ValueError(
'Unexpected type %(expected_type)s'
% {'expected_type': type(expected)})
return matched_value
def project_id(self):
# We'll allow test cases to override the default project id. This is
# useful when using multiple tenants.
project_id = None
project_id = self.api.project_id
except AttributeError:
return project_id or PROJECT_ID
def project_id(self, project_id):
self.api.project_id = project_id
# Reset cached credentials
self.api.auth_result = None
def generalize_subs(self, subs, vanilla_regexes):
"""Give the test a chance to modify subs after the server response
was verified, and before the on-disk doc/api_samples file is checked.
This may be needed by some tests to convert exact matches expected
from the server into pattern matches to verify what is in the
sample file.
If there are no changes to be made, subs is returned unharmed.
return subs
def _update_links(self, sample_data):
"""Process sample data and update version specific links."""
# replace version urls
project_id_exp = '(%s|%s)' % (PROJECT_ID, self.project_id)
url_re = self._get_host() + r"/v(2|2\.1)/" + project_id_exp
new_url = self._get_host() + "/" + self.api_major_version
new_url += "/" + self.project_id
updated_data = re.sub(url_re, new_url, sample_data)
# replace unversioned urls
url_re = self._get_host() + "/" + project_id_exp
new_url = self._get_host()
new_url += "/" + self.project_id
updated_data = re.sub(url_re, new_url, updated_data)
return updated_data
def _verify_response(self, name, subs, response, exp_code,
# Always also include the laundry list of base regular
# expressions for possible key values in our templates. Test
# specific patterns (the value of ``subs``) can override
# these.
regexes = self._get_regexes()
subs = regexes
self.subs = subs
message = response.text if response.status_code >= 400 else None
self.assertEqual(exp_code, response.status_code, message)
response_data = response.content
response_data = pretty_data(response_data)
if not os.path.exists(self._get_template(name,
self._write_template(name, response_data)
template_data = response_data
template_data = self._read_template(name)
if (self.generate_samples and
not os.path.exists(self._get_sample(
name, self.microversion))):
self._write_sample(name, response_data)
sample_data = response_data
with open(self._get_sample(name,
self.microversion)) as sample:
sample_data =
if update_links:
sample_data = self._update_links(sample_data)
template_data = objectify(template_data)
response_data = objectify(response_data)
response_result = self._compare_result(template_data,
response_data, "Response")
except NoMatch as e:
raise NoMatch("\nFailed to match Template to Response: \n%s\n"
"Template: %s\n\n"
"Response: %s\n\n" %
# NOTE(danms): replace some of the subs with patterns for the
# doc/api_samples check, which won't have things like the
# correct compute host name. Also let the test do some of its
# own generalization, if necessary
vanilla_regexes = self._get_regexes()
subs['compute_host'] = vanilla_regexes['host_name']
subs['id'] = vanilla_regexes['id']
subs['uuid'] = vanilla_regexes['uuid']
subs['image_id'] = vanilla_regexes['uuid']
subs = self.generalize_subs(subs, vanilla_regexes)
self.subs = subs
sample_data = objectify(sample_data)
self._compare_result(template_data, sample_data, "Sample")
return response_result
except NoMatch as e:
raise NoMatch("\nFailed to match Template to Sample: \n%s\n"
"Template: %s\n\n"
"Sample: %s\n\n"
"Hint: does your test need to override "
"ApiSampleTestBase.generalize_subs()?" %
def _get_host(self):
return ''
def _get_glance_host(self):
return ''
def _get_regexes(self):
text = r'(\\"|[^"])*'
isotime_re = r'\d{4}-[0,1]\d-[0-3]\dT\d{2}:\d{2}:\d{2}Z'
strtime_re = r'\d{4}-[0,1]\d-[0-3]\dT\d{2}:\d{2}:\d{2}\.\d{6}'
strtime_url_re = (r'\d{4}-[0,1]\d-[0-3]\d'
xmltime_re = (r'\d{4}-[0,1]\d-[0-3]\d '
# NOTE(claudiub): the x509 keypairs are different from the
# ssh keypairs. For example, the x509 fingerprint has 40 bytes.
return {
'isotime': isotime_re,
'strtime': strtime_re,
'strtime_url': strtime_url_re,
'strtime_or_none': r'None|%s' % strtime_re,
'xmltime': xmltime_re,
'password': '[0-9a-zA-Z]{1,12}',
'ip': '[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}.[0-9]{1,3}',
'ip6': '([0-9a-zA-Z]{1,4}:){1,7}:?[0-9a-zA-Z]{1,4}',
'id': '(?P<id>[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}'
'uuid': '[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}'
'request_id': 'req-[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}'
'reservation_id': 'r-[0-9a-zA-Z]{8}',
'private_key': '(-----BEGIN RSA PRIVATE KEY-----|)'
'(-----END RSA PRIVATE KEY-----|)',
'public_key': '(ssh-rsa|-----BEGIN CERTIFICATE-----)'
'[ a-zA-Z0-9\n/+=]*'
'(Generated-by-Nova|-----END CERTIFICATE-----)',
'fingerprint': '(([0-9a-f]{2}:){19}|([0-9a-f]{2}:){15})'
'keypair_type': 'ssh|x509',
'host': self._get_host(),
'host_name': r'\w+',
'glance_host': self._get_glance_host(),
'text': text,
'int': '[0-9]+',
'user_id': text,
'api_vers': self.api_major_version,
'compute_endpoint': self._get_compute_endpoint(),
'versioned_compute_endpoint': self._get_vers_compute_endpoint(),
def _get_compute_endpoint(self):
# NOTE(sdague): "openstack" is stand in for project_id, it
# should be more generic in future.
return '%s/%s' % (self._get_host(), self.project_id)
return self._get_host()
def _get_vers_compute_endpoint(self):
# NOTE(sdague): "openstack" is stand in for project_id, it
# should be more generic in future.
return '%s/%s/%s' % (self._get_host(), self.api_major_version,
return '%s/%s' % (self._get_host(), self.api_major_version)
def _get_response(self, url, method, body=None, strip_version=False,
headers = headers or {}
headers['Content-Type'] = 'application/json'
headers['Accept'] = 'application/json'
return self.api.api_request(url, body=body, method=method,
headers=headers, strip_version=strip_version)
def _do_options(self, url, strip_version=False, headers=None):
return self._get_response(url, 'OPTIONS', strip_version=strip_version,
def _do_get(self, url, strip_version=False, headers=None,
response = self._get_response(url, 'GET', strip_version=strip_version,
if return_json_body and hasattr(response, 'content'):
return jsonutils.loads(response.content)
return response
def _do_post(self, url, name=None, subs=None, method='POST', headers=None):
self.subs = {} if subs is None else subs
body = None
if name:
body = self._read_template(name) % self.subs
sample = self._get_sample(name, self.microversion)
if self.generate_samples and not os.path.exists(sample):
self._write_sample(name, body)
return self._get_response(url, method, body, headers=headers)
def _do_put(self, url, name=None, subs=None, headers=None):
# name indicates that we have a body document. While the HTTP
# spec implies that PUT is supposed to have one, we have some
# APIs which don't.
if name:
return self._do_post(
url, name, subs, method='PUT', headers=headers)
return self._get_response(url, 'PUT', headers=headers)
def _do_delete(self, url, headers=None):
return self._get_response(url, 'DELETE', headers=headers)