192 lines
6.7 KiB
Python
192 lines
6.7 KiB
Python
# 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 atexit
|
|
import os
|
|
import time
|
|
|
|
import fixtures
|
|
import nose.plugins.attrib
|
|
import testresources
|
|
import testtools
|
|
|
|
from tempest import clients
|
|
from tempest import config
|
|
from tempest.openstack.common import log as logging
|
|
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
# All the successful HTTP status codes from RFC 2616
|
|
HTTP_SUCCESS = (200, 201, 202, 203, 204, 205, 206)
|
|
|
|
|
|
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
|
|
|
|
|
|
# there is a mis-match between nose and testtools for older pythons.
|
|
# testtools will set skipException to be either
|
|
# unittest.case.SkipTest, unittest2.case.SkipTest or an internal skip
|
|
# exception, depending on what it can find. Python <2.7 doesn't have
|
|
# unittest.case.SkipTest; so if unittest2 is not installed it falls
|
|
# back to the internal class.
|
|
#
|
|
# The current nose skip plugin will decide to raise either
|
|
# unittest.case.SkipTest or its own internal exception; it does not
|
|
# look for unittest2 or the internal unittest exception. Thus we must
|
|
# monkey-patch testtools.TestCase.skipException to be the exception
|
|
# the nose skip plugin expects.
|
|
#
|
|
# However, with the switch to testr nose may not be available, so we
|
|
# require you to opt-in to this fix with an environment variable.
|
|
#
|
|
# This is temporary until upstream nose starts looking for unittest2
|
|
# as testtools does; we can then remove this and ensure unittest2 is
|
|
# available for older pythons; then nose and testtools will agree
|
|
# unittest2.case.SkipTest is the one-true skip test exception.
|
|
#
|
|
# https://review.openstack.org/#/c/33056
|
|
# https://github.com/nose-devs/nose/pull/699
|
|
if 'TEMPEST_PY26_NOSE_COMPAT' in os.environ:
|
|
try:
|
|
import unittest.case.SkipTest
|
|
# convince pep8 we're using the import...
|
|
if unittest.case.SkipTest:
|
|
pass
|
|
raise RuntimeError("You have unittest.case.SkipTest; "
|
|
"no need to override")
|
|
except ImportError:
|
|
LOG.info("Overriding skipException to nose SkipTest")
|
|
testtools.TestCase.skipException = nose.plugins.skip.SkipTest
|
|
|
|
at_exit_set = set()
|
|
|
|
|
|
def validate_tearDownClass():
|
|
if at_exit_set:
|
|
raise RuntimeError("tearDownClass does not calls the super's "
|
|
"tearDownClass in these classes: "
|
|
+ str(at_exit_set))
|
|
|
|
atexit.register(validate_tearDownClass)
|
|
|
|
|
|
class BaseTestCase(testtools.TestCase,
|
|
testtools.testcase.WithAttributes,
|
|
testresources.ResourcedTestCase):
|
|
|
|
config = config.TempestConfig()
|
|
|
|
setUpClassCalled = False
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
if hasattr(super(BaseTestCase, cls), 'setUpClass'):
|
|
super(BaseTestCase, cls).setUpClass()
|
|
cls.setUpClassCalled = True
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
at_exit_set.remove(cls)
|
|
if hasattr(super(BaseTestCase, cls), 'tearDownClass'):
|
|
super(BaseTestCase, cls).tearDownClass()
|
|
|
|
def setUp(self):
|
|
super(BaseTestCase, self).setUp()
|
|
if not self.setUpClassCalled:
|
|
raise RuntimeError("setUpClass does not calls the super's"
|
|
"setUpClass in the "
|
|
+ self.__class__.__name__)
|
|
at_exit_set.add(self.__class__)
|
|
test_timeout = os.environ.get('OS_TEST_TIMEOUT', 0)
|
|
try:
|
|
test_timeout = int(test_timeout)
|
|
except ValueError:
|
|
test_timeout = 0
|
|
if test_timeout > 0:
|
|
self.useFixture(fixtures.Timeout(test_timeout, gentle=True))
|
|
|
|
if (os.environ.get('OS_STDOUT_CAPTURE') == 'True' or
|
|
os.environ.get('OS_STDOUT_CAPTURE') == '1'):
|
|
stdout = self.useFixture(fixtures.StringStream('stdout')).stream
|
|
self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
|
|
if (os.environ.get('OS_STDERR_CAPTURE') == 'True' or
|
|
os.environ.get('OS_STDERR_CAPTURE') == '1'):
|
|
stderr = self.useFixture(fixtures.StringStream('stderr')).stream
|
|
self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
|
|
|
|
@classmethod
|
|
def _get_identity_admin_client(cls):
|
|
"""
|
|
Returns an instance of the Identity Admin API client
|
|
"""
|
|
os = clients.AdminManager(interface=cls._interface)
|
|
admin_client = os.identity_client
|
|
return admin_client
|
|
|
|
@classmethod
|
|
def _get_client_args(cls):
|
|
|
|
return (
|
|
cls.config,
|
|
cls.config.identity.admin_username,
|
|
cls.config.identity.admin_password,
|
|
cls.config.identity.uri
|
|
)
|
|
|
|
|
|
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
|