#!/usr/bin/env python

import os

import sys

import re

import gc

import locale

import shutil

import time

import unittest

import doctest

import operator

import subprocess

import tempfile

import traceback

import warnings

try:

import platform

IS_PYPY = platform.python_implementation() == 'PyPy'

IS_CPYTHON = platform.python_implementation() == 'CPython'

except (ImportError, AttributeError):

IS_CPYTHON = True

IS_PYPY = False

try:

from StringIO import StringIO

except ImportError:

from io import StringIO

try:

import cPickle as pickle

except ImportError:

import pickle

try:

from io import open as io_open

except ImportError:

from codecs import open as io_open

try:

import threading

except ImportError: # No threads, no problems

threading = None

try:

from collections import defaultdict

except ImportError:

class defaultdict(object):

def __init__(self, default_factory=lambda : None):

self._dict = {}

self.default_factory = default_factory

def __getitem__(self, key):

if key not in self._dict:

self._dict[key] = self.default_factory()

return self._dict[key]

def __setitem__(self, key, value):

self._dict[key] = value

def __contains__(self, key):

return key in self._dict

def __repr__(self):

return repr(self._dict)

def __nonzero__(self):

return bool(self._dict)

try:

basestring

except NameError:

basestring = str

WITH_CYTHON = True

CY3_DIR = None

from distutils.dist import Distribution

from distutils.core import Extension

from distutils.command.build_ext import build_ext as _build_ext

from distutils import sysconfig

distutils_distro = Distribution()

if sys.platform == 'win32':

# TODO: Figure out why this hackery (see http://thread.gmane.org/gmane.comp.python.cython.devel/8280/).

config_files = distutils_distro.find_config_files()

try: config_files.remove('setup.cfg')

except ValueError: pass

distutils_distro.parse_config_files(config_files)

cfgfiles = distutils_distro.find_config_files()

try: cfgfiles.remove('setup.cfg')

except ValueError: pass

distutils_distro.parse_config_files(cfgfiles)

EXT_DEP_MODULES = {

'tag:numpy' : 'numpy',

'tag:pstats': 'pstats',

'tag:posix' : 'posix',

'tag:array' : 'array',

}

def patch_inspect_isfunction():

import inspect

orig_isfunction = inspect.isfunction

def isfunction(obj):

return orig_isfunction(obj) or type(obj).__name__ == 'cython_function_or_method'

isfunction._orig_isfunction = orig_isfunction

inspect.isfunction = isfunction

def unpatch_inspect_isfunction():

import inspect

try:

orig_isfunction = inspect.isfunction._orig_isfunction

except AttributeError:

pass

else:

inspect.isfunction = orig_isfunction

def update_linetrace_extension(ext):

ext.define_macros.append(('CYTHON_TRACE', 1))

return ext

def update_numpy_extension(ext):

import numpy

from numpy.distutils.misc_util import get_info

ext.include_dirs.append(numpy.get_include())

# We need the npymath library for numpy.math.

# This is typically a static-only library.

for attr, value in get_info('npymath').items():

getattr(ext, attr).extend(value)

def update_openmp_extension(ext):

ext.openmp = True

language = ext.language

if language == 'cpp':

flags = OPENMP_CPP_COMPILER_FLAGS

else:

flags = OPENMP_C_COMPILER_FLAGS

if flags:

compile_flags, link_flags = flags

ext.extra_compile_args.extend(compile_flags.split())

ext.extra_link_args.extend(link_flags.split())

return ext

elif sys.platform == 'win32':

return ext

return EXCLUDE_EXT

def get_openmp_compiler_flags(language):

"""

As of gcc 4.2, it supports OpenMP 2.5. Gcc 4.4 implements 3.0. We don't

(currently) check for other compilers.

returns a two-tuple of (CFLAGS, LDFLAGS) to build the OpenMP extension

"""

if language == 'cpp':

cc = sysconfig.get_config_var('CXX')

else:

cc = sysconfig.get_config_var('CC')

if not cc:

if sys.platform == 'win32':

return '/openmp', ''

return None

# For some reason, cc can be e.g. 'gcc -pthread'

cc = cc.split()[0]

# Force english output

env = os.environ.copy()

env['LC_MESSAGES'] = 'C'

matcher = re.compile(r"gcc version (\d+\.\d+)").search

try:

p = subprocess.Popen([cc, "-v"], stderr=subprocess.PIPE, env=env)

except EnvironmentError:

# Be compatible with Python 3

warnings.warn("Unable to find the %s compiler: %s: %s" %

(language, os.strerror(sys.exc_info()[1].errno), cc))

return None

_, output = p.communicate()

output = output.decode(locale.getpreferredencoding() or 'ASCII', 'replace')

gcc_version = matcher(output)

if not gcc_version:

return None # not gcc - FIXME: do something about other compilers

compiler_version = gcc_version.group(1)

if compiler_version and compiler_version.split('.') >= ['4', '2']:

return '-fopenmp', '-fopenmp'

try:

locale.setlocale(locale.LC_ALL, '')

except locale.Error:

pass

OPENMP_C_COMPILER_FLAGS = get_openmp_compiler_flags('c')

OPENMP_CPP_COMPILER_FLAGS = get_openmp_compiler_flags('cpp')

# Return this from the EXT_EXTRAS matcher callback to exclude the extension

EXCLUDE_EXT = object()

EXT_EXTRAS = {

'tag:numpy' : update_numpy_extension,

'tag:openmp': update_openmp_extension,

'tag:trace' : update_linetrace_extension,

}

def _is_py3_before_32(excluded, version):

return version[0] >= 3 and version < (3,2)

# TODO: use tags

VER_DEP_MODULES = {

# tests are excluded if 'CurrentPythonVersion OP VersionTuple', i.e.

# (2,4) : (operator.lt, ...) excludes ... when PyVer < 2.4.x

(2,4) : (operator.lt, lambda x: x in ['run.extern_builtins_T258',

'run.builtin_sorted',

'run.reversed_iteration',

]),

(2,5) : (operator.lt, lambda x: x in ['run.any',

'run.all',

'run.yield_from_pep380', # GeneratorExit

'run.generator_frame_cycle', # yield in try-finally

'run.generator_expressions_in_class',

'run.absolute_import',

'run.relativeimport_T542',

'run.relativeimport_star_T542',

'run.initial_file_path', # relative import

'run.pynumber_subtype_conversion', # bug in Py2.4

'build.cythonize_script', # python2.4 -m a.b.c

'build.cythonize_script_excludes', # python2.4 -m a.b.c

'build.cythonize_script_package', # python2.4 -m a.b.c

]),

(2,6) : (operator.lt, lambda x: x in ['run.print_function',

'run.language_level', # print function

'run.cython3',

'run.property_decorator_T593', # prop.setter etc.

'run.generators_py', # generators, with statement

'run.pure_py', # decorators, with statement

'run.purecdef',

'run.struct_conversion',

'run.bytearray_coercion',

'run.bytearraymethods',

'run.bytearray_ascii_auto_encoding',

'run.bytearray_default_auto_encoding',

# memory views require buffer protocol

'memoryview.relaxed_strides',

'memoryview.cythonarray',

'memoryview.memslice',

'memoryview.numpy_memoryview',

'memoryview.memoryviewattrs',

'memoryview.memoryview',

'run.withstat_py',

]),

(2,7) : (operator.lt, lambda x: x in ['run.withstat_py27', # multi context with statement

'run.yield_inside_lambda',

'run.test_dictviews',

'run.pyclass_special_methods',

]),

# The next line should start (3,); but this is a dictionary, so

# we can only have one (3,) key. Since 2.7 is supposed to be the

# last 2.x release, things would have to change drastically for this

# to be unsafe...

(2,999): (operator.lt, lambda x: x in ['run.special_methods_T561_py3',

'run.test_raisefrom',

]),

(3,): (operator.ge, lambda x: x in ['run.non_future_division',

'compile.extsetslice',

'compile.extdelslice',

'run.special_methods_T561_py2'

]),

(3,1): (_is_py3_before_32, lambda x: x in ['run.pyclass_special_methods',

]),

(3,3) : (operator.lt, lambda x: x in ['build.package_compilation',

]),

(3,4,0,'beta',3) : (operator.le, lambda x: x in ['run.py34_signature',

]),

}

# files that should not be converted to Python 3 code with 2to3

KEEP_2X_FILES = [

os.path.join('Cython', 'Debugger', 'Tests', 'test_libcython_in_gdb.py'),

os.path.join('Cython', 'Debugger', 'Tests', 'test_libpython_in_gdb.py'),

os.path.join('Cython', 'Debugger', 'libcython.py'),

os.path.join('Cython', 'Debugger', 'libpython.py'),

]

COMPILER = None

INCLUDE_DIRS = [ d for d in os.getenv('INCLUDE', '').split(os.pathsep) if d ]

CFLAGS = os.getenv('CFLAGS', '').split()

CCACHE = os.getenv('CYTHON_RUNTESTS_CCACHE', '').split()

TEST_SUPPORT_DIR = 'testsupport'

BACKENDS = ['c', 'cpp']

UTF8_BOM_BYTES = r'\xef\xbb\xbf'.encode('ISO-8859-1').decode('unicode_escape')

def memoize(f):

uncomputed = object()

f._cache = {}

def func(*args):

res = f._cache.get(args, uncomputed)

if res is uncomputed:

res = f._cache[args] = f(*args)

return res

return func

@memoize

def parse_tags(filepath):

tags = defaultdict(list)

parse_tag = re.compile(r'#\s*(\w+)\s*:(.*)$').match

f = io_open(filepath, encoding='ISO-8859-1', errors='ignore')

try:

for line in f:

# ignore BOM-like bytes and whitespace

line = line.lstrip(UTF8_BOM_BYTES).strip()

if not line:

if tags:

break # assume all tags are in one block

else:

continue

if line[0] != '#':

break

parsed = parse_tag(line)

if parsed:

tag, values = parsed.groups()

if tag in ('coding', 'encoding'):

continue

if tag == 'tags':

tag = 'tag'

print("WARNING: test tags use the 'tag' directive, not 'tags' (%s)" % filepath)

if tag not in ('mode', 'tag', 'ticket', 'cython', 'distutils'):

print("WARNING: unknown test directive '%s' found (%s)" % (tag, filepath))

values = values.split(',')

tags[tag].extend(filter(None, [value.strip() for value in values]))

elif tags:

break # assume all tags are in one block

finally:

f.close()

return tags

list_unchanging_dir = memoize(lambda x: os.listdir(x))

@memoize

def _list_pyregr_data_files(test_directory):

is_data_file = re.compile('(?:[.](txt|pem|db|html)|^bad.*[.]py)$').search

return ['__init__.py'] + [

filename for filename in list_unchanging_dir(test_directory)

if is_data_file(filename)]

def import_ext(module_name, file_path=None):

if file_path:

import imp

return imp.load_dynamic(module_name, file_path)

else:

try:

from importlib import invalidate_caches

except ImportError:

pass

else:

invalidate_caches()

return __import__(module_name, globals(), locals(), ['*'])

class build_ext(_build_ext):

def build_extension(self, ext):

try:

try: # Py2.7+ & Py3.2+

compiler_obj = self.compiler_obj

except AttributeError:

compiler_obj = self.compiler

if ext.language == 'c++':

compiler_obj.compiler_so.remove('-Wstrict-prototypes')

if CCACHE:

compiler_obj.compiler_so = CCACHE + compiler_obj.compiler_so

if getattr(ext, 'openmp', None) and compiler_obj.compiler_type == 'msvc':

ext.extra_compile_args.append('/openmp')

except Exception:

pass

_build_ext.build_extension(self, ext)

class ErrorWriter(object):

match_error = re.compile('(warning:)?(?:.*:)?\s*([-0-9]+)\s*:\s*([-0-9]+)\s*:\s*(.*)').match

def __init__(self):

self.output = []

self.write = self.output.append

def _collect(self, collect_errors, collect_warnings):

s = ''.join(self.output)

result = []

for line in s.split('\n'):

match = self.match_error(line)

if match:

is_warning, line, column, message = match.groups()

if (is_warning and collect_warnings) or \

(not is_warning and collect_errors):

result.append( (int(line), int(column), message.strip()) )

result.sort()

return [ "%d:%d: %s" % values for values in result ]

def geterrors(self):

return self._collect(True, False)

def getwarnings(self):

return self._collect(False, True)

def getall(self):

return self._collect(True, True)

class TestBuilder(object):

def __init__(self, rootdir, workdir, selectors, exclude_selectors, annotate,

cleanup_workdir, cleanup_sharedlibs, cleanup_failures,

with_pyregr, cython_only, languages, test_bugs, fork, language_level):

self.rootdir = rootdir

self.workdir = workdir

self.selectors = selectors

self.exclude_selectors = exclude_selectors

self.annotate = annotate

self.cleanup_workdir = cleanup_workdir

self.cleanup_sharedlibs = cleanup_sharedlibs

self.cleanup_failures = cleanup_failures

self.with_pyregr = with_pyregr

self.cython_only = cython_only

self.languages = languages

self.test_bugs = test_bugs

self.fork = fork

self.language_level = language_level

def build_suite(self):

suite = unittest.TestSuite()

filenames = os.listdir(self.rootdir)

filenames.sort()

for filename in filenames:

path = os.path.join(self.rootdir, filename)

if os.path.isdir(path) and filename != TEST_SUPPORT_DIR:

if filename == 'pyregr' and not self.with_pyregr:

continue

if filename == 'broken' and not self.test_bugs:

continue

suite.addTest(

self.handle_directory(path, filename))

if sys.platform not in ['win32']:

# Non-Windows makefile.

if [1 for selector in self.selectors if selector("embedded")] \

and not [1 for selector in self.exclude_selectors if selector("embedded")]:

suite.addTest(unittest.makeSuite(EmbedTest))

return suite

def handle_directory(self, path, context):

workdir = os.path.join(self.workdir, context)

if not os.path.exists(workdir):

os.makedirs(workdir)

suite = unittest.TestSuite()

filenames = list_unchanging_dir(path)

filenames.sort()

for filename in filenames:

filepath = os.path.join(path, filename)

module, ext = os.path.splitext(filename)

if ext not in ('.py', '.pyx', '.srctree'):

continue

if filename.startswith('.'):

continue # certain emacs backup files

if context == 'pyregr':

tags = defaultdict(list)

else:

tags = parse_tags(filepath)

fqmodule = "%s.%s" % (context, module)

if not [ 1 for match in self.selectors

if match(fqmodule, tags) ]:

continue

if self.exclude_selectors:

if [1 for match in self.exclude_selectors

if match(fqmodule, tags)]:

continue

mode = 'run' # default

if tags['mode']:

mode = tags['mode'][0]

elif context == 'pyregr':

mode = 'pyregr'

if ext == '.srctree':

if 'cpp' not in tags['tag'] or 'cpp' in self.languages:

suite.addTest(EndToEndTest(filepath, workdir, self.cleanup_workdir))

continue

# Choose the test suite.

if mode == 'pyregr':

if not filename.startswith('test_'):

continue

test_class = CythonPyregrTestCase

elif mode == 'run':

if module.startswith("test_"):

test_class = CythonUnitTestCase

else:

test_class = CythonRunTestCase

else:

test_class = CythonCompileTestCase

for test in self.build_tests(test_class, path, workdir,

module, mode == 'error', tags):

suite.addTest(test)

if mode == 'run' and ext == '.py' and not self.cython_only:

# additionally test file in real Python

suite.addTest(PureDoctestTestCase(module, os.path.join(path, filename)))

return suite

def build_tests(self, test_class, path, workdir, module, expect_errors, tags):

if 'werror' in tags['tag']:

warning_errors = True

else:

warning_errors = False

if expect_errors:

if 'cpp' in tags['tag'] and 'cpp' in self.languages:

languages = ['cpp']

else:

languages = self.languages[:1]

else:

languages = self.languages

if 'cpp' in tags['tag'] and 'c' in languages:

languages = list(languages)

languages.remove('c')

elif 'no-cpp' in tags['tag'] and 'cpp' in self.languages:

languages = list(languages)

languages.remove('cpp')

tests = [ self.build_test(test_class, path, workdir, module, tags,

language, expect_errors, warning_errors)

for language in languages ]

return tests

def build_test(self, test_class, path, workdir, module, tags,

language, expect_errors, warning_errors):

language_workdir = os.path.join(workdir, language)

if not os.path.exists(language_workdir):

os.makedirs(language_workdir)

workdir = os.path.join(language_workdir, module)

return test_class(path, workdir, module, tags,

language=language,

expect_errors=expect_errors,

annotate=self.annotate,

cleanup_workdir=self.cleanup_workdir,

cleanup_sharedlibs=self.cleanup_sharedlibs,

cleanup_failures=self.cleanup_failures,

cython_only=self.cython_only,

fork=self.fork,

language_level=self.language_level,

warning_errors=warning_errors)

class CythonCompileTestCase(unittest.TestCase):

def __init__(self, test_directory, workdir, module, tags, language='c',

expect_errors=False, annotate=False, cleanup_workdir=True,

cleanup_sharedlibs=True, cleanup_failures=True, cython_only=False,

fork=True, language_level=2, warning_errors=False):

self.test_directory = test_directory

self.tags = tags

self.workdir = workdir

self.module = module

self.language = language

self.expect_errors = expect_errors

self.annotate = annotate

self.cleanup_workdir = cleanup_workdir

self.cleanup_sharedlibs = cleanup_sharedlibs

self.cleanup_failures = cleanup_failures

self.cython_only = cython_only

self.fork = fork

self.language_level = language_level

self.warning_errors = warning_errors

unittest.TestCase.__init__(self)

def shortDescription(self):

return "compiling (%s) %s" % (self.language, self.module)

def setUp(self):

from Cython.Compiler import Options

self._saved_options = [ (name, getattr(Options, name))

for name in ('warning_errors',

'clear_to_none',

'error_on_unknown_names',

'error_on_uninitialized') ]

self._saved_default_directives = Options.directive_defaults.items()

Options.warning_errors = self.warning_errors

if sys.version_info >= (3, 4):

Options.directive_defaults['autotestdict'] = False

if not os.path.exists(self.workdir):

os.makedirs(self.workdir)

if self.workdir not in sys.path:

sys.path.insert(0, self.workdir)

def tearDown(self):

from Cython.Compiler import Options

for name, value in self._saved_options:

setattr(Options, name, value)

Options.directive_defaults = dict(self._saved_default_directives)

unpatch_inspect_isfunction()

try:

sys.path.remove(self.workdir)

except ValueError:

pass

try:

del sys.modules[self.module]

except KeyError:

pass

cleanup = self.cleanup_failures or self.success

cleanup_c_files = WITH_CYTHON and self.cleanup_workdir and cleanup

cleanup_lib_files = self.cleanup_sharedlibs and cleanup

if os.path.exists(self.workdir):

if cleanup_c_files and cleanup_lib_files:

shutil.rmtree(self.workdir, ignore_errors=True)

else:

for rmfile in os.listdir(self.workdir):

if not cleanup_c_files:

if (rmfile[-2:] in (".c", ".h") or

rmfile[-4:] == ".cpp" or

rmfile.endswith(".html") and rmfile.startswith(self.module)):

continue

if not cleanup_lib_files and (rmfile.endswith(".so") or rmfile.endswith(".dll")):

continue

try:

rmfile = os.path.join(self.workdir, rmfile)

if os.path.isdir(rmfile):

shutil.rmtree(rmfile, ignore_errors=True)

else:

os.remove(rmfile)

except IOError:

pass

def runTest(self):

self.success = False

self.runCompileTest()

self.success = True

def runCompileTest(self):

return self.compile(

self.test_directory, self.module, self.workdir,

self.test_directory, self.expect_errors, self.annotate)

def find_module_source_file(self, source_file):

if not os.path.exists(source_file):

source_file = source_file[:-1]

return source_file

def build_target_filename(self, module_name):

target = '%s.%s' % (module_name, self.language)

return target

def related_files(self, test_directory, module_name):

is_related = re.compile('%s_.*[.].*' % module_name).match

return [filename for filename in list_unchanging_dir(test_directory)

if is_related(filename)]

def copy_files(self, test_directory, target_directory, file_list):

# use symlink on Unix, copy on Windows

try:

copy = os.symlink

except AttributeError:

copy = shutil.copy

join = os.path.join

for filename in file_list:

file_path = join(test_directory, filename)

if os.path.exists(file_path):

copy(file_path, join(target_directory, filename))

def source_files(self, workdir, module_name, file_list):

return ([self.build_target_filename(module_name)] +

[filename for filename in file_list

if not os.path.isfile(os.path.join(workdir, filename))])

def split_source_and_output(self, test_directory, module, workdir):

source_file = self.find_module_source_file(os.path.join(test_directory, module) + '.pyx')

source_and_output = io_open(source_file, 'rU', encoding='ISO-8859-1')

try:

out = io_open(os.path.join(workdir, module + os.path.splitext(source_file)[1]),

'w', encoding='ISO-8859-1')

for line in source_and_output:

if line.startswith("_ERRORS"):

out.close()

out = ErrorWriter()

else:

out.write(line)

finally:

source_and_output.close()

try:

geterrors = out.geterrors

except AttributeError:

out.close()

return []

else:

return geterrors()

def run_cython(self, test_directory, module, targetdir, incdir, annotate,

extra_compile_options=None):

include_dirs = INCLUDE_DIRS + [os.path.join(test_directory, '..', TEST_SUPPORT_DIR)]

if incdir:

include_dirs.append(incdir)

source = self.find_module_source_file(

os.path.join(test_directory, module + '.pyx'))

target = os.path.join(targetdir, self.build_target_filename(module))

if extra_compile_options is None:

extra_compile_options = {}

try:

CompilationOptions

except NameError:

from Cython.Compiler.Main import CompilationOptions

from Cython.Compiler.Main import compile as cython_compile

from Cython.Compiler.Main import default_options

options = CompilationOptions(

default_options,

include_path = include_dirs,

output_file = target,

annotate = annotate,

use_listing_file = False,

cplus = self.language == 'cpp',

language_level = self.language_level,

generate_pxi = False,

evaluate_tree_assertions = True,

**extra_compile_options

)

cython_compile(source, options=options,

full_module_name=module)

def run_distutils(self, test_directory, module, workdir, incdir,

extra_extension_args=None):

cwd = os.getcwd()

os.chdir(workdir)

try:

build_extension = build_ext(distutils_distro)

build_extension.include_dirs = INCLUDE_DIRS[:]

if incdir:

build_extension.include_dirs.append(incdir)

build_extension.finalize_options()

if COMPILER:

build_extension.compiler = COMPILER

ext_compile_flags = CFLAGS[:]

compiler = COMPILER or sysconfig.get_config_var('CC')

if self.language == 'c' and compiler == 'gcc':

ext_compile_flags.extend(['-std=c89', '-pedantic'])

if build_extension.compiler == 'mingw32':

ext_compile_flags.append('-Wno-format')

if extra_extension_args is None:

extra_extension_args = {}

related_files = self.related_files(test_directory, module)

self.copy_files(test_directory, workdir, related_files)

extension = Extension(

module,

sources = self.source_files(workdir, module, related_files),

extra_compile_args = ext_compile_flags,

**extra_extension_args

)

if self.language == 'cpp':

# Set the language now as the fixer might need it

extension.language = 'c++'

if 'distutils' in self.tags:

from Cython.Build.Dependencies import DistutilsInfo

pyx_path = os.path.join(self.test_directory, self.module + ".pyx")

DistutilsInfo(open(pyx_path)).apply(extension)

for matcher, fixer in list(EXT_EXTRAS.items()):

if isinstance(matcher, str):

# lazy init

del EXT_EXTRAS[matcher]

matcher = string_selector(matcher)

EXT_EXTRAS[matcher] = fixer

if matcher(module, self.tags):

newext = fixer(extension)

if newext is EXCLUDE_EXT:

return

extension = newext or extension

if self.language == 'cpp':

extension.language = 'c++'

build_extension.extensions = [extension]

build_extension.build_temp = workdir

build_extension.build_lib = workdir

build_extension.run()

finally:

os.chdir(cwd)

try:

get_ext_fullpath = build_extension.get_ext_fullpath

except AttributeError:

def get_ext_fullpath(ext_name, self=build_extension):

# copied from distutils.command.build_ext (missing in Py2.[45])

fullname = self.get_ext_fullname(ext_name)

modpath = fullname.split('.')

filename = self.get_ext_filename(modpath[-1])

if not self.inplace:

filename = os.path.join(*modpath[:-1]+[filename])

return os.path.join(self.build_lib, filename)

package = '.'.join(modpath[0:-1])

build_py = self.get_finalized_command('build_py')

package_dir = os.path.abspath(build_py.get_package_dir(package))

return os.path.join(package_dir, filename)

return get_ext_fullpath(module)

def compile(self, test_directory, module, workdir, incdir,

expect_errors, annotate):

expected_errors = errors = ()

if expect_errors:

expected_errors = self.split_source_and_output(

test_directory, module, workdir)

test_directory = workdir

if WITH_CYTHON:

old_stderr = sys.stderr

try:

sys.stderr = ErrorWriter()

self.run_cython(test_directory, module, workdir, incdir, annotate)

errors = sys.stderr.geterrors()

finally:

sys.stderr = old_stderr

if errors or expected_errors:

try:

for expected, error in zip(expected_errors, errors):

self.assertEquals(expected, error)

if len(errors) < len(expected_errors):

expected_error = expected_errors[len(errors)]

self.assertEquals(expected_error, None)

elif len(errors) > len(expected_errors):

unexpected_error = errors[len(expected_errors)]

self.assertEquals(None, unexpected_error)

except AssertionError:

print("\n=== Expected errors: ===")

print('\n'.join(expected_errors))

print("\n\n=== Got errors: ===")

print('\n'.join(errors))

print('\n')

raise

return None

if self.cython_only:

so_path = None

else:

so_path = self.run_distutils(test_directory, module, workdir, incdir)

return so_path

class CythonRunTestCase(CythonCompileTestCase):

def setUp(self):

CythonCompileTestCase.setUp(self)

from Cython.Compiler import Options

Options.clear_to_none = False

def shortDescription(self):

if self.cython_only:

return CythonCompileTestCase.shortDescription(self)

else:

return "compiling (%s) and running %s" % (self.language, self.module)

def run(self, result=None):

if result is None:

result = self.defaultTestResult()

result.startTest(self)

try:

self.setUp()

try:

self.success = False

ext_so_path = self.runCompileTest()

failures, errors = len(result.failures), len(result.errors)

if not self.cython_only:

self.run_tests(result, ext_so_path)

if failures == len(result.failures) and errors == len(result.errors):

# No new errors...

self.success = True

finally:

check_thread_termination()

except Exception:

result.addError(self, sys.exc_info())

result.stopTest(self)

try:

self.tearDown()

except Exception:

pass

def run_tests(self, result, ext_so_path):

self.run_doctests(self.module, result, ext_so_path)

def run_doctests(self, module_or_name, result, ext_so_path):

def run_test(result):

if isinstance(module_or_name, basestring):

module = import_ext(module_or_name, ext_so_path)

else:

module = module_or_name

tests = doctest.DocTestSuite(module)

tests.run(result)

run_forked_test(result, run_test, self.shortDescription(), self.fork)

def run_forked_test(result, run_func, test_name, fork=True):

if not fork or sys.version_info[0] >= 3 or not hasattr(os, 'fork'):

run_func(result)

sys.stdout.flush()

sys.stderr.flush()

gc.collect()

return

# fork to make sure we do not keep the tested module loaded

result_handle, result_file = tempfile.mkstemp()

os.close(result_handle)

child_id = os.fork()

if not child_id:

result_code = 0

output = None

try:

try:

tests = partial_result = None

try:

partial_result = PartialTestResult(result)

run_func(partial_result)

sys.stdout.flush()

sys.stderr.flush()

gc.collect()

except Exception:

result_code = 1

if partial_result is not None:

if tests is None:

# importing failed, try to fake a test class

tests = _FakeClass(

failureException=sys.exc_info()[1],

_shortDescription=test_name,

module_name=None)

partial_result.addError(tests, sys.exc_info())

output = open(result_file, 'wb')

pickle.dump(partial_result.data(), output)

except:

traceback.print_exc()

finally:

try: sys.stderr.flush()

except: pass

try: sys.stdout.flush()

except: pass

try:

if output is not None:

output.close()

except:

pass

os._exit(result_code)

try:

cid, result_code = os.waitpid(child_id, 0)

module_name = test_name.split()[-1]

# os.waitpid returns the child's result code in the

# upper byte of result_code, and the signal it was

# killed by in the lower byte

if result_code & 255:

raise Exception("Tests in module '%s' were unexpectedly killed by signal %d"%

(module_name, result_code & 255))

result_code >>= 8

if result_code in (0,1):

input = open(result_file, 'rb')

try:

PartialTestResult.join_results(result, pickle.load(input))

finally:

input.close()

if result_code:

raise Exception("Tests in module '%s' exited with status %d" %

(module_name, result_code))

finally:

try: os.unlink(result_file)

except: pass

class PureDoctestTestCase(unittest.TestCase):

def __init__(self, module_name, module_path):

self.module_name = module_name

self.module_path = module_path

unittest.TestCase.__init__(self, 'run')

def shortDescription(self):

return "running pure doctests in %s" % self.module_name

def run(self, result=None):

if result is None:

result = self.defaultTestResult()

loaded_module_name = 'pure_doctest__' + self.module_name

result.startTest(self)

try: