def rmtree(path):
"""Remove the given recursively.
:note: we use shutil rmtree but adjust its behaviour to see whether files that
couldn't be deleted are read-only. Windows will not remove them in that case"""
def onerror(func, path, exc_info):
# Is the error an access error ?
os.chmod(path, stat.S_IWUSR)
try:
func(path) # Will scream if still not possible to delete.
except Exception as ex:
if HIDE_WINDOWS_KNOWN_ERRORS:
raise SkipTest("FIXME: fails with: PermissionError\n %s", ex)
else:
raise
return shutil.rmtree(path, False, onerror)
python类SkipTest()的实例源码
def addOnException(self, exception_handler):
@wraps(exception_handler)
def wrapped_handler(exc_info):
if issubclass(exc_info[0], SkipTest):
return
return exception_handler(exc_info)
super(BaseTestCase, self).addOnException(wrapped_handler)
def options(self, parser, env):
"""
Add my options to command line.
"""
env_opt = 'NOSE_WITHOUT_SKIP'
parser.add_option('--no-skip', action='store_true',
dest='noSkip', default=env.get(env_opt, False),
help="Disable special handling of SkipTest "
"exceptions.")
def test_upgrade(self):
raise SkipTest("upgrade memcached")
# add tests for touch, cas, gets etc
def test_unserializable_span_with_finish():
try:
import numpy as np
except ImportError:
raise SkipTest("numpy not installed")
# a weird case where manually calling finish with an unserializable
# span was causing an loop of serialization.
writer = DummyWriter()
tracer = Tracer()
tracer.writer = writer
with tracer.trace("parent") as span:
span.metrics['as'] = np.int64(1) # circumvent the data checks
span.finish()
def test_set_numpy_metric():
try:
import numpy as np
except ImportError:
raise SkipTest("numpy not installed")
s = Span(tracer=None, name="test.span")
s.set_metric("a", np.int64(1))
eq_(s.get_metric("a"), 1)
eq_(type(s.get_metric("a")), float)
def options(self, parser, env):
"""
Add my options to command line.
"""
env_opt = 'NOSE_WITHOUT_SKIP'
parser.add_option('--no-skip', action='store_true',
dest='noSkip', default=env.get(env_opt, False),
help="Disable special handling of SkipTest "
"exceptions.")
def start_fixture(self):
if os.environ.get('GABBI_SKIP_NETWORK', 'False').lower() == 'true':
raise case.SkipTest('live tests skipped')
def start_fixture(self):
raise case.SkipTest('entire suite skipped')
def is_extension_enabled(extension_name, service):
"""A function that will check the list of enabled extensions from config
"""
config_dict = {
'compute': CONF.compute_feature_enabled.api_extensions,
'compute_v3': CONF.compute_feature_enabled.api_v3_extensions,
'volume': CONF.volume_feature_enabled.api_extensions,
'network': CONF.network_feature_enabled.api_extensions,
'object': CONF.object_storage_feature_enabled.discoverable_apis,
}
if config_dict[service][0] == 'all':
return True
if extension_name in config_dict[service]:
return True
return False
# 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
def assertSetup():
try:
Client.guess_config()
except RuntimeError:
raise SkipTest("no default config is detected")
def start_fixture(self):
"""Set up config."""
global LOAD_APP_KWARGS
self.conf = None
# Determine the database connection.
db_url = os.environ.get('PIFPAF_URL', "sqlite://").replace(
"mysql://", "mysql+pymysql://")
if not db_url:
raise case.SkipTest('No database connection configured')
conf = self.conf = service.prepare_service([], [])
opts.set_defaults(self.conf)
content = ('{"default": ""}')
if six.PY3:
content = content.encode('utf-8')
self.tempfile = fileutils.write_to_tempfile(content=content,
prefix='policy',
suffix='.json')
conf.set_override("policy_file", self.tempfile,
group='oslo_policy')
conf.set_override(
'api_paste_config',
os.path.abspath('etc/panko/api_paste.ini')
)
parsed_url = list(urlparse.urlparse(db_url))
parsed_url[2] += '-%s' % uuidutils.generate_uuid(dashed=False)
db_url = urlparse.urlunparse(parsed_url)
conf.set_override('connection', db_url, group='database')
if (parsed_url[0].startswith("mysql")
or parsed_url[0].startswith("postgresql")):
sqlalchemy_utils.create_database(conf.database.connection)
self.conn = storage.get_connection_from_config(self.conf)
self.conn.upgrade()
LOAD_APP_KWARGS = {
'conf': conf, 'appname': 'panko+noauth',
}