/usr/lib/python2.7/dist-packages/tempest/test.py is in python-tempest 2013.2.a1291.g23a1b4f-0ubuntu2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | # vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2012 OpenStack, LLC
# 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 time
import nose.plugins.attrib
import testresources
import testtools
from tempest.common import log as logging
from tempest import config
from tempest import manager
LOG = logging.getLogger(__name__)
def attr(*args, **kwargs):
"""A decorator which applies the nose and testtools attr decorator
This decorator applies the nose attr decorator as well as the
the testtools.testcase.attr if it is in the list of attributes
to testtools we want to apply.
"""
def decorator(f):
if 'type' in kwargs and isinstance(kwargs['type'], str):
f = testtools.testcase.attr(kwargs['type'])(f)
if kwargs['type'] == 'smoke':
f = testtools.testcase.attr('gate')(f)
elif 'type' in kwargs and isinstance(kwargs['type'], list):
for attr in kwargs['type']:
f = testtools.testcase.attr(attr)(f)
if attr == 'smoke':
f = testtools.testcase.attr('gate')(f)
return nose.plugins.attrib.attr(*args, **kwargs)(f)
return decorator
class BaseTestCase(testtools.TestCase,
testtools.testcase.WithAttributes,
testresources.ResourcedTestCase):
config = config.TempestConfig()
@classmethod
def setUpClass(cls):
if hasattr(super(BaseTestCase, cls), 'setUpClass'):
super(BaseTestCase, cls).setUpClass()
def call_until_true(func, duration, sleep_for):
"""
Call the given function until it returns True (and return True) or
until the specified duration (in seconds) elapses (and return
False).
:param func: A zero argument callable that returns True on success.
:param duration: The number of seconds for which to attempt a
successful call of the function.
:param sleep_for: The number of seconds to sleep after an unsuccessful
invocation of the function.
"""
now = time.time()
timeout = now + duration
while now < timeout:
if func():
return True
LOG.debug("Sleeping for %d seconds", sleep_for)
time.sleep(sleep_for)
now = time.time()
return False
class TestCase(BaseTestCase):
"""Base test case class for all Tempest tests
Contains basic setup and convenience methods
"""
manager_class = None
@classmethod
def setUpClass(cls):
cls.manager = cls.manager_class()
for attr_name in cls.manager.client_attr_names:
# Ensure that pre-existing class attributes won't be
# accidentally overriden.
assert not hasattr(cls, attr_name)
client = getattr(cls.manager, attr_name)
setattr(cls, attr_name, client)
cls.resource_keys = {}
cls.os_resources = []
def set_resource(self, key, thing):
LOG.debug("Adding %r to shared resources of %s" %
(thing, self.__class__.__name__))
self.resource_keys[key] = thing
self.os_resources.append(thing)
def get_resource(self, key):
return self.resource_keys[key]
def remove_resource(self, key):
thing = self.resource_keys[key]
self.os_resources.remove(thing)
del self.resource_keys[key]
def status_timeout(self, things, thing_id, expected_status):
"""
Given a thing and an expected status, do a loop, sleeping
for a configurable amount of time, checking for the
expected status to show. At any time, if the returned
status of the thing is ERROR, fail out.
"""
def check_status():
# python-novaclient has resources available to its client
# that all implement a get() method taking an identifier
# for the singular resource to retrieve.
thing = things.get(thing_id)
new_status = thing.status
if new_status == 'ERROR':
self.fail("%s failed to get to expected status."
"In ERROR state."
% thing)
elif new_status == expected_status:
return True # All good.
LOG.debug("Waiting for %s to get to %s status. "
"Currently in %s status",
thing, expected_status, new_status)
conf = config.TempestConfig()
if not call_until_true(check_status,
conf.compute.build_timeout,
conf.compute.build_interval):
self.fail("Timed out waiting for thing %s to become %s"
% (thing_id, expected_status))
class ComputeFuzzClientTest(TestCase):
"""
Base test case class for OpenStack Compute API (Nova)
that uses the Tempest REST fuzz client libs for calling the API.
"""
manager_class = manager.ComputeFuzzClientManager
|