"verbose", "use_resources", "max_memuse", "record_original_stdout",

"get_original_stdout", "unload", "unlink", "rmtree", "forget",

"is_resource_enabled", "requires", "find_unused_port", "bind_port",

"fcmp", "have_unicode", "is_jython", "TESTFN", "HOST", "FUZZ",

"SAVEDCWD", "temp_cwd", "findfile", "sortdict", "check_syntax_error",

"open_urlresource", "check_warnings", "check_py3k_warnings",

"CleanImport", "EnvironmentVarGuard", "captured_output",

"captured_stdout", "TransientResource", "transient_internet",

"run_with_locale", "set_memlimit", "bigmemtest", "bigaddrspacetest",

"BasicTestRunner", "run_unittest", "run_doctest", "threading_setup",

"threading_cleanup", "reap_children", "cpython_only",

"check_impl_detail", "get_attribute", "py3k_bytes",

"""Test skipped because it requested a disallowed resource.

"""Context manager to suppress package and module deprecation

if ignore:

with warnings.catch_warnings():

warnings.filterwarnings("ignore", ".+ (module|package)",

DeprecationWarning)

yield

else:

"""Import and return the module to be tested, raising SkipTest if

with _ignore_deprecated_imports(deprecated):

try:

return importlib.import_module(name)

except ImportError, msg:

"""Helper function to save and remove a module from sys.modules

# try to import the module and raise an error if it can't be imported

if name not in sys.modules:

__import__(name)

del sys.modules[name]

for modname in list(sys.modules):

if modname == name or modname.startswith(name + '.'):

orig_modules[modname] = sys.modules[modname]

"""Helper function to save and block a module in sys.modules

saved = True

try:

orig_modules[name] = sys.modules[name]

except KeyError:

saved = False

sys.modules[name] = None

"""Imports and returns a module, deliberately bypassing the sys.modules cache

# NOTE: test_heapq, test_json, and test_warnings include extra sanity

# checks to make sure that this utility function is working as expected

with _ignore_deprecated_imports(deprecated):

# Keep track of modules saved for later restoration as well

# as those which just need a blocking entry removed

orig_modules = {}

names_to_remove = []

_save_and_remove_module(name, orig_modules)

try:

for fresh_name in fresh:

_save_and_remove_module(fresh_name, orig_modules)

for blocked_name in blocked:

if not _save_and_block_module(blocked_name, orig_modules):

names_to_remove.append(blocked_name)

fresh_module = importlib.import_module(name)

except ImportError:

fresh_module = None

finally:

for orig_name, module in orig_modules.items():

sys.modules[orig_name] = module

for name_to_remove in names_to_remove:

del sys.modules[name_to_remove]

"""Get an attribute, raising SkipTest if AttributeError is raised."""

try:

attribute = getattr(obj, name)

except AttributeError:

raise unittest.SkipTest("module %s has no attribute %s" % (

obj.__name__, name))

else:

try:

del sys.modules[name]

except KeyError:

try:

os.unlink(filename)

except OSError:

try:

shutil.rmtree(path)

except OSError, e:

# Unix returns ENOENT, Windows returns ESRCH.

if e.errno not in (errno.ENOENT, errno.ESRCH):

'''"Forget" a module was ever imported by removing it from sys.modules and

unload(modname)

for dirname in sys.path:

unlink(os.path.join(dirname, modname + os.extsep + 'pyc'))

# Deleting the .pyo file cannot be within the 'try' for the .pyc since

# the chance exists that there is no .pyc (and thus the 'try' statement

# is exited) but there is a .pyo file.

"""Raise ResourceDenied if the specified resource is not available.

# see if the caller's module is __main__ - if so, treat as if

# the resource was set

if sys._getframe(1).f_globals.get("__name__") == "__main__":

return

if not is_resource_enabled(resource):

if msg is None:

msg = "Use of the `%s' resource not enabled" % resource

"""Returns an unused port that should be suitable for binding. This is

tempsock = socket.socket(family, socktype)

port = bind_port(tempsock)

tempsock.close()

del tempsock

"""Bind the socket to a free port and return the port number. Relies on

if sock.family == socket.AF_INET and sock.type == socket.SOCK_STREAM:

if hasattr(socket, 'SO_REUSEADDR'):

if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR) == 1:

raise TestFailed("tests should never set the SO_REUSEADDR " \

"socket option on TCP/IP sockets!")

if hasattr(socket, 'SO_REUSEPORT'):

if sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT) == 1:

raise TestFailed("tests should never set the SO_REUSEPORT " \

"socket option on TCP/IP sockets!")

if hasattr(socket, 'SO_EXCLUSIVEADDRUSE'):

sock.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)

sock.bind((host, 0))

port = sock.getsockname()[1]

if isinstance(x, float) or isinstance(y, float):

try:

fuzz = (abs(x) + abs(y)) * FUZZ

if abs(x-y) <= fuzz:

return 0

except:

pass

elif type(x) == type(y) and isinstance(x, (tuple, list)):

for i in range(min(len(x), len(y))):

outcome = fcmp(x[i], y[i])

if outcome != 0:

return outcome

return (len(x) > len(y)) - (len(x) < len(y))

"""

if isinstance(name, unicode):

try:

name = name.encode(sys.getfilesystemencoding() or 'ascii')

except UnicodeEncodeError:

if not quiet:

raise unittest.SkipTest('unable to encode the cwd name with '

'the filesystem encoding.')

saved_dir = os.getcwd()

is_temporary = False

try:

os.mkdir(name)

os.chdir(name)

is_temporary = True

except OSError:

if not quiet:

raise

warnings.warn('tests may fail, unable to change the CWD to ' + name,

RuntimeWarning, stacklevel=3)

try:

yield os.getcwd()

finally:

os.chdir(saved_dir)

if is_temporary:

"""Try to find a file on sys.path and the working directory. If it is not

if os.path.isabs(file):

return file

if subdir is not None:

file = os.path.join(subdir, file)

path = sys.path

path = [os.path.dirname(here)] + path

for dn in path:

fn = os.path.join(dn, file)

if os.path.exists(fn): return fn

"Like repr(dict), but in sorted order."

items = dict.items()

items.sort()

reprpairs = ["%r: %r" % pair for pair in items]

withcommas = ", ".join(reprpairs)

"""

file = open(TESTFN, "wb")

try:

return file.fileno()

finally:

file.close()

import urlparse, urllib2

filename = urlparse.urlparse(url)[2].split('/')[-1] # '/': it's URL!

fn = os.path.join(os.path.dirname(__file__), "data", filename)

def check_valid_file(fn):

f = open(fn)

if check is None:

return f

elif check(f):

f.seek(0)

return f

f.close()

if os.path.exists(fn):

f = check_valid_file(fn)

if f is not None:

return f

unlink(fn)

# Verify the requirement before downloading the file

requires('urlfetch')

print >> get_original_stdout(), '\tfetching %s ...' % url

f = urllib2.urlopen(url, timeout=15)

try:

with open(fn, "wb") as out:

s = f.read()

while s:

out.write(s)

s = f.read()

finally:

f.close()

f = check_valid_file(fn)

if f is not None:

return f

"""Convenience wrapper for the warnings list returned on

def __init__(self, warnings_list):

self._warnings = warnings_list

self._last = 0

def __getattr__(self, attr):

if len(self._warnings) > self._last:

return getattr(self._warnings[-1], attr)

elif attr in warnings.WarningMessage._WARNING_DETAILS:

return None

raise AttributeError("%r has no attribute %r" % (self, attr))

@property

def warnings(self):

return self._warnings[self._last:]

def reset(self):

"""Catch the warnings, then check if all the expected

# Clear the warning registry of the calling module

# in order to re-raise the warnings.

frame = sys._getframe(2)

registry = frame.f_globals.get('__warningregistry__')

if registry:

registry.clear()

with warnings.catch_warnings(record=True) as w:

# Set filter "always" to record all warnings. Because

# test_warnings swap the module, we need to look up in

# the sys.modules dictionary.

sys.modules['warnings'].simplefilter("always")

yield WarningsRecorder(w)

# Filter the recorded warnings

reraise = [warning.message for warning in w]

missing = []

for msg, cat in filters:

seen = False

for exc in reraise[:]:

message = str(exc)

# Filter out the matching messages

if (re.match(msg, message, re.I) and

issubclass(exc.__class__, cat)):

seen = True

reraise.remove(exc)

if not seen and not quiet:

# This filter caught nothing

missing.append((msg, cat.__name__))

if reraise:

raise AssertionError("unhandled warning %r" % reraise[0])

if missing:

raise AssertionError("filter (%r, %s) did not catch any warning" %

"""Context manager to silence warnings.

quiet = kwargs.get('quiet')

if not filters:

filters = (("", Warning),)

# Preserve backward compatibility

if quiet is None:

quiet = True

"""Context manager to silence py3k warnings.

if sys.py3kwarning:

if not filters:

filters = (("", DeprecationWarning),)

else:

# It should not raise any py3k warning

filters = ()

"""Context manager to force import to return a new module reference.

def __init__(self, *module_names):

self.original_modules = sys.modules.copy()

for module_name in module_names:

if module_name in sys.modules:

module = sys.modules[module_name]

# It is possible that module_name is just an alias for

# another module (e.g. stub for modules renamed in 3.x).

# In that case, we also need delete the real module to clear

# the import cache.

if module.__name__ != module_name:

del sys.modules[module.__name__]

del sys.modules[module_name]

def __enter__(self):

return self

def __exit__(self, *ignore_exc):

"""Class to help protect the environment variable properly. Can be used as

def __init__(self):

self._environ = os.environ

self._changed = {}

def __getitem__(self, envvar):

return self._environ[envvar]

def __setitem__(self, envvar, value):

# Remember the initial value on the first access

if envvar not in self._changed:

self._changed[envvar] = self._environ.get(envvar)

self._environ[envvar] = value

def __delitem__(self, envvar):

# Remember the initial value on the first access

if envvar not in self._changed:

self._changed[envvar] = self._environ.get(envvar)

if envvar in self._environ:

del self._environ[envvar]

def keys(self):

return self._environ.keys()

def set(self, envvar, value):

self[envvar] = value

def unset(self, envvar):

del self[envvar]

def __enter__(self):

return self

def __exit__(self, *ignore_exc):

for (k, v) in self._changed.items():

if v is None:

if k in self._environ:

del self._environ[k]

else:

self._environ[k] = v

"""Context manager to temporarily add directories to sys.path.

def __init__(self, *paths):

self.original_value = sys.path[:]

self.original_object = sys.path

sys.path.extend(paths)

def __enter__(self):

return self

def __exit__(self, *ignore_exc):

sys.path = self.original_object

"""Raise ResourceDenied if an exception is raised while the context manager

def __init__(self, exc, **kwargs):

self.exc = exc

self.attrs = kwargs

def __enter__(self):

return self

def __exit__(self, type_=None, value=None, traceback=None):

"""If type_ is a subclass of self.exc and value has attributes matching

if type_ is not None and issubclass(self.exc, type_):

for attr, attr_value in self.attrs.iteritems():

if not hasattr(value, attr):

break

if getattr(value, attr) != attr_value:

break

else:

"""Return a context manager that raises ResourceDenied when various issues

default_errnos = [

('ECONNREFUSED', 111),

('ECONNRESET', 104),

('EHOSTUNREACH', 113),

('ENETUNREACH', 101),

('ETIMEDOUT', 110),

]

default_gai_errnos = [

('EAI_NONAME', -2),

('EAI_NODATA', -5),

]

denied = ResourceDenied("Resource '%s' is not available" % resource_name)

captured_errnos = errnos

gai_errnos = []

if not captured_errnos:

captured_errnos = [getattr(errno, name, num)

for (name, num) in default_errnos]

gai_errnos = [getattr(socket, name, num)

for (name, num) in default_gai_errnos]

def filter_error(err):

n = getattr(err, 'errno', None)

if (isinstance(err, socket.timeout) or

(isinstance(err, socket.gaierror) and n in gai_errnos) or

n in captured_errnos):

if not verbose:

sys.stderr.write(denied.args[0] + "\n")

raise denied

old_timeout = socket.getdefaulttimeout()

try:

if timeout is not None:

socket.setdefaulttimeout(timeout)

yield

except IOError as err:

# urllib can wrap original socket errors multiple times (!), we must

# unwrap to get at the original error.

while True:

a = err.args

if len(a) >= 1 and isinstance(a[0], IOError):

err = a[0]

# The error can also be wrapped as args[1]:

# except socket.error as msg:

# raise IOError('socket error', msg).with_traceback(sys.exc_info()[2])

elif len(a) >= 2 and isinstance(a[1], IOError):

err = a[1]

else:

break

filter_error(err)

raise

# XXX should we catch generic exceptions and look for their

# __cause__ or __context__?

finally:

"""Return a context manager used by captured_stdout and captured_stdin

import StringIO

orig_stdout = getattr(sys, stream_name)

setattr(sys, stream_name, StringIO.StringIO())

try:

yield getattr(sys, stream_name)

finally:

"""Capture the output of sys.stdout:

"""Force as many objects as possible to be collected.

gc.collect()

if is_jython:

time.sleep(0.1)

gc.collect()

def decorator(func):

def inner(*args, **kwds):

try:

import locale

category = getattr(locale, catstr)

orig_locale = locale.setlocale(category)

except AttributeError:

# if the test author gives us an invalid category string

raise

except:

# cannot retrieve original locale, so do nothing

locale = orig_locale = None

else:

for loc in locales:

try:

locale.setlocale(category, loc)

break

except:

pass

# now run the function, resetting the locale on exceptions

try:

return func(*args, **kwds)

finally:

if locale and orig_locale:

locale.setlocale(category, orig_locale)

inner.func_name = func.func_name

inner.__doc__ = func.__doc__

return inner

global max_memuse

global real_max_memuse

sizes = {

'k': 1024,

'm': _1M,

'g': _1G,

't': 1024*_1G,

}

m = re.match(r'(\d+(\.\d+)?) (K|M|G|T)b?$', limit,

re.IGNORECASE | re.VERBOSE)

if m is None:

raise ValueError('Invalid memory limit %r' % (limit,))

memlimit = int(float(m.group(1)) * sizes[m.group(3).lower()])

real_max_memuse = memlimit

if memlimit > MAX_Py_ssize_t:

memlimit = MAX_Py_ssize_t

if memlimit < _2G - 1:

raise ValueError('Memory limit %r too low to be useful' % (limit,))

"""Decorator for bigmem tests.

def decorator(f):

def wrapper(self):

if not max_memuse:

# If max_memuse is 0 (the default),

# we still want to run the tests with size set to a few kb,

# to make sure they work. We still want to avoid using

# too much memory, though, but we do that noisily.

maxsize = 5147

self.assertFalse(maxsize * memuse + overhead > 20 * _1M)

else:

maxsize = int((max_memuse - overhead) / memuse)

if maxsize < minsize:

# Really ought to print 'test skipped' or something

if verbose:

sys.stderr.write("Skipping %s because of memory "

"constraint\n" % (f.__name__,))

return

# Try to keep some breathing room in memory use

maxsize = max(maxsize - 50 * _1M, minsize)

return f(self, maxsize)

wrapper.minsize = minsize

wrapper.memuse = memuse

wrapper.overhead = overhead

return wrapper

def decorator(f):

def wrapper(self):

if not real_max_memuse:

maxsize = 5147

else:

maxsize = size

if real_max_memuse and real_max_memuse < maxsize * memuse:

if verbose:

sys.stderr.write("Skipping %s because of memory "

"constraint\n" % (f.__name__,))

return

return f(self, maxsize)

wrapper.size = size

wrapper.memuse = memuse

wrapper.overhead = overhead

return wrapper

"""Decorator for tests that fill the address space."""

def wrapper(self):

if max_memuse < MAX_Py_ssize_t:

if verbose:

sys.stderr.write("Skipping %s because of memory "

"constraint\n" % (f.__name__,))

else:

return f(self)

def run(self, test):