runtests.py

#!/usr/bin/env python

from __future__ import print_function

import atexit
import base64
import os
import sys
import re
import gc
import heapq
import locale
import shutil
import time
import unittest
import doctest
import operator
import subprocess
import tempfile
import traceback
import warnings
import zlib
import glob
from contextlib import contextmanager
from collections import defaultdict

try:
    import platform
    IS_PYPY = platform.python_implementation() == 'PyPy'
    IS_CPYTHON = platform.python_implementation() == 'CPython'
except (ImportError, AttributeError):
    IS_CPYTHON = True
    IS_PYPY = False
IS_PY2 = sys.version_info[0] < 3

from io import open as io_open
try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO  # doesn't accept 'str' in Py2

try:
    import cPickle as pickle
except ImportError:
    import pickle

try:
    import threading
except ImportError: # No threads, no problems
    threading = None

try:
    from unittest import SkipTest
except ImportError:
    class SkipTest(Exception):  # don't raise, only provided to allow except-ing it!
        pass
    def skip_test(reason):
        sys.stderr.write("Skipping test: %s\n" % reason)
else:
    def skip_test(reason):
        raise SkipTest(reason)

try:
    basestring
except NameError:
    basestring = str

WITH_CYTHON = True

from distutils.command.build_ext import build_ext as _build_ext
from distutils import sysconfig
_to_clean = []

@atexit.register
def _cleanup_files():
    """
    This is only used on Cygwin to clean up shared libraries that are unsafe
    to delete while the test suite is running.
    """

    for filename in _to_clean:
        if os.path.isdir(filename):
            shutil.rmtree(filename, ignore_errors=True)
        else:
            try:
                os.remove(filename)
            except OSError:
                pass


def get_distutils_distro(_cache=[]):
    if _cache:
        return _cache[0]
    # late import to accommodate for setuptools override
    from distutils.dist import Distribution
    distutils_distro = Distribution()

    if sys.platform == 'win32':
        # TODO: Figure out why this hackery (see https://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)
    _cache.append(distutils_distro)
    return distutils_distro


EXT_DEP_MODULES = {
    'tag:numpy':     'numpy',
    'tag:pythran':  'pythran',
    'tag:setuptools':  'setuptools.sandbox',
    'tag:asyncio':  'asyncio',
    'tag:pstats':   'pstats',
    'tag:posix':    'posix',
    'tag:array':    'array',
    'tag:coverage': 'Cython.Coverage',
    'Coverage':     'Cython.Coverage',
    'tag:ipython':  'IPython.testing.globalipapp',
    'tag:jedi':     'jedi_BROKEN_AND_DISABLED',
    'tag:test.support': 'test.support',  # support module for CPython unit tests
}

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 def_to_cdef(source):
    '''
    Converts the module-level def methods into cdef methods, i.e.

        @decorator
        def foo([args]):
            """
            [tests]
            """
            [body]

    becomes

        def foo([args]):
            """
            [tests]
            """
            return foo_c([args])

        cdef foo_c([args]):
            [body]
    '''
    output = []
    skip = False
    def_node = re.compile(r'def (\w+)\(([^()*]*)\):').match
    lines = iter(source.split('\n'))
    for line in lines:
        if not line.strip():
            output.append(line)
            continue

        if skip:
            if line[0] != ' ':
                skip = False
            else:
                continue

        if line[0] == '@':
            skip = True
            continue

        m = def_node(line)
        if m:
            name = m.group(1)
            args = m.group(2)
            if args:
                args_no_types = ", ".join(arg.split()[-1] for arg in args.split(','))
            else:
                args_no_types = ""
            output.append("def %s(%s):" % (name, args_no_types))
            line = next(lines)
            if '"""' in line:
                has_docstring = True
                output.append(line)
                for line in lines:
                    output.append(line)
                    if '"""' in line:
                        break
            else:
                has_docstring = False
            output.append("    return %s_c(%s)" % (name, args_no_types))
            output.append('')
            output.append("cdef %s_c(%s):" % (name, args))
            if not has_docstring:
                output.append(line)

        else:
            output.append(line)

    return '\n'.join(output)


def exclude_extension_in_pyver(*versions):
    def check(ext):
        return EXCLUDE_EXT if sys.version_info[:2] in versions else ext
    return check


def exclude_extension_on_platform(*platforms):
    def check(ext):
        return EXCLUDE_EXT if sys.platform in platforms else ext
    return check


def update_linetrace_extension(ext):
    ext.define_macros.append(('CYTHON_TRACE', 1))
    return ext


def update_numpy_extension(ext, set_api17_macro=True):
    import numpy
    from numpy.distutils.misc_util import get_info

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

    if set_api17_macro:
        ext.define_macros.append(('NPY_NO_DEPRECATED_API', 'NPY_1_7_API_VERSION'))

    # 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 sys.platform == 'win32' and sys.version_info[:2] == (3,4):
        # OpenMP tests fail in appveyor in Py3.4 -> just ignore them, EoL of Py3.4 is early 2019...
        return EXCLUDE_EXT

    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 update_cpp11_extension(ext):
    """
        update cpp11 extensions that will run on versions of gcc >4.8
    """
    gcc_version = get_gcc_version(ext.language)
    if gcc_version:
        compiler_version = gcc_version.group(1)
        if float(compiler_version) > 4.8:
            ext.extra_compile_args.append("-std=c++11")
        return ext

    clang_version = get_clang_version(ext.language)
    if clang_version:
        ext.extra_compile_args.append("-std=c++11")
        if sys.platform == "darwin":
          ext.extra_compile_args.append("-stdlib=libc++")
          ext.extra_compile_args.append("-mmacosx-version-min=10.7")
        return ext

    return EXCLUDE_EXT

def update_pthread_extension(ext):
    """
      use pthread to link and compile on platform which supports it
    """
    if sys.platform in ('linux', 'linux2', 'freebsd', 'netbsd', 'openbsd'):
        # FIXME: this check on the platform is hacky
        ext.extra_compile_args.append("-pthread")
        ext.extra_link_args.append("-pthread")

def get_cc_version(language):
    """
        finds gcc version using Popen
    """
    if language == 'cpp':
        cc = sysconfig.get_config_var('CXX')
    else:
        cc = sysconfig.get_config_var('CC')
    if not cc:
        from distutils import ccompiler
        cc = ccompiler.get_default_compiler()

    if not cc:
        return ''

    # 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'
    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 ''
    _, output = p.communicate()
    return output.decode(locale.getpreferredencoding() or 'ASCII', 'replace')


def get_gcc_version(language):
    matcher = re.compile(r"gcc version (\d+\.\d+)").search
    return matcher(get_cc_version(language))


def get_clang_version(language):
    matcher = re.compile(r"clang(?:-|\s+version\s+)(\d+\.\d+)").search
    return matcher(get_cc_version(language))


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
    """
    gcc_version = get_gcc_version(language)

    if not gcc_version:
        if sys.platform == 'win32':
            return '/openmp', ''
        else:
            return None # not gcc - FIXME: do something about other compilers

    # gcc defines "__int128_t", assume that at least all 64 bit architectures have it
    global COMPILER_HAS_INT128
    COMPILER_HAS_INT128 = getattr(sys, 'maxsize', getattr(sys, 'maxint', 0)) > 2**60

    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

COMPILER = None
COMPILER_HAS_INT128 = False
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:cpp11': update_cpp11_extension,
    'tag:trace' : update_linetrace_extension,
    'tag:pthread': update_pthread_extension,
    'tag:bytesformat':  exclude_extension_in_pyver((3, 3), (3, 4)),  # no %-bytes formatting
    'tag:no-macos':  exclude_extension_on_platform('darwin'),
    'tag:py3only':  exclude_extension_in_pyver((2, 7)),
}


# 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

    # 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',
                                           'run.different_package_names',
                                           'run.unicode_imports',  # encoding problems on appveyor in Py2
                                           'run.reimport_failure',  # reimports don't do anything in Py2
                                           ]),
    (3,): (operator.ge, lambda x: x in ['run.non_future_division',
                                        'compile.extsetslice',
                                        'compile.extdelslice',
                                        'run.special_methods_T561_py2',
                                        ]),
    (3,3) : (operator.lt, lambda x: x in ['build.package_compilation',
                                          'build.cythonize_pep420_namespace',
                                          'run.yield_from_py33',
                                          'pyximport.pyximport_namespace',
                                          'run.qualname',
                                          ]),
    (3,4): (operator.lt, lambda x: x in ['run.py34_signature',
                                         'run.test_unicode',  # taken from Py3.7, difficult to backport
                                         ]),
    (3,4,999): (operator.gt, lambda x: x in ['run.initial_file_path',
                                             ]),
    (3,5): (operator.lt, lambda x: x in ['run.py35_pep492_interop',
                                         'run.py35_asyncio_async_def',
                                         'run.mod__spec__',
                                         'run.pep526_variable_annotations',  # typing module
                                         'run.test_exceptions',  # copied from Py3.7+
                                         ]),
}

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
    with io_open(filepath, encoding='ISO-8859-1', errors='ignore') as f:
        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', 'preparse'):
                    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
    return tags


list_unchanging_dir = memoize(lambda x: os.listdir(x))  # needs lambda to set function attribute


@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(r'(warning:)?(?:.*:)?\s*([-0-9]+)\s*:\s*([-0-9]+)\s*:\s*(.*)').match

    def __init__(self, encoding=None):
        self.output = []
        self.encoding = encoding

    def write(self, value):
        if self.encoding:
            value = value.encode('ISO-8859-1').decode(self.encoding)
        self.output.append(value)

    def _collect(self):
        s = ''.join(self.output)
        results = {'errors': [], 'warnings': []}
        for line in s.splitlines():
            match = self.match_error(line)
            if match:
                is_warning, line, column, message = match.groups()
                results['warnings' if is_warning else 'errors'].append((int(line), int(column), message.strip()))

        return [["%d:%d: %s" % values for values in sorted(results[key])] for key in ('errors', 'warnings')]

    def geterrors(self):
        return self._collect()[0]

    def getwarnings(self):
        return self._collect()[1]

    def getall(self):
        return self._collect()

    def close(self):
        pass  # ignore, only to match file-like interface


class Stats(object):
    def __init__(self, top_n=8):
        self.top_n = top_n
        self.test_counts = defaultdict(int)
        self.test_times = defaultdict(float)
        self.top_tests = defaultdict(list)

    def add_time(self, name, language, metric, t):
        self.test_counts[metric] += 1
        self.test_times[metric] += t
        top = self.top_tests[metric]
        push = heapq.heappushpop if len(top) >= self.top_n else heapq.heappush
        # min-heap => pop smallest/shortest until longest times remain
        push(top, (t, name, language))

    @contextmanager
    def time(self, name, language, metric):
        t = time.time()
        yield
        t = time.time() - t
        self.add_time(name, language, metric, t)

    def update(self, stats):
        # type: (Stats) -> None
        for metric, t in stats.test_times.items():
            self.test_times[metric] += t
            self.test_counts[metric] += stats.test_counts[metric]
            top = self.top_tests[metric]
            for entry in stats.top_tests[metric]:
                push = heapq.heappushpop if len(top) >= self.top_n else heapq.heappush
                push(top, entry)

    def print_stats(self, out=sys.stderr):
        if not self.test_times:
            return
        lines = ['Times:\n']
        for metric, t in sorted(self.test_times.items()):
            count = self.test_counts[metric]
            top = self.top_tests[metric]
            lines.append("%-12s: %8.2f sec  (%4d, %6.3f / run) - slowest: %s\n" % (
                metric, t, count, t / count,
                ', '.join("'{2}:{1}' ({0:.2f}s)".format(*item) for item in heapq.nlargest(self.top_n, top))))
        out.write(''.join(lines))


class TestBuilder(object):
    def __init__(self, rootdir, workdir, selectors, exclude_selectors, options,
                 with_pyregr, languages, test_bugs, language_level,
                 common_utility_dir, pythran_dir=None,
                 default_mode='run', stats=None,
                 add_embedded_test=False):
        self.rootdir = rootdir
        self.workdir = workdir
        self.selectors = selectors
        self.exclude_selectors = exclude_selectors
        self.annotate = options.annotate_source
        self.cleanup_workdir = options.cleanup_workdir
        self.cleanup_sharedlibs = options.cleanup_sharedlibs
        self.cleanup_failures = options.cleanup_failures
        self.with_pyregr = with_pyregr
        self.cython_only = options.cython_only
        self.doctest_selector = re.compile(options.only_pattern).search if options.only_pattern else None
        self.languages = languages
        self.test_bugs = test_bugs
        self.fork = options.fork
        self.language_level = language_level
        self.test_determinism = options.test_determinism
        self.common_utility_dir = common_utility_dir
        self.pythran_dir = pythran_dir
        self.default_mode = default_mode
        self.stats = stats
        self.add_embedded_test = add_embedded_test
        self.capture = options.capture

    def build_suite(self):
        suite = unittest.TestSuite()
        filenames = os.listdir(self.rootdir)
        filenames.sort()
        # TODO: parallelise I/O with a thread pool for the different directories once we drop Py2 support
        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'] and self.add_embedded_test:
            # 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 = self.default_mode
            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, stats=self.stats,
                             capture=self.capture))
                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
            elif mode in ['compile', 'error']:
                test_class = CythonCompileTestCase
            else:
                raise KeyError('Invalid test mode: ' + mode)

            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 and not filename.startswith('test_'):
                # additionally test file in real Python
                min_py_ver = [
                    (int(pyver.group(1)), int(pyver.group(2)))
                    for pyver in map(re.compile(r'pure([0-9]+)[.]([0-9]+)').match, tags['tag'])
                    if pyver
                ]
                if not min_py_ver or any(sys.version_info >= min_ver for min_ver in min_py_ver):
                    suite.addTest(PureDoctestTestCase(module, os.path.join(path, filename), tags, stats=self.stats))

        return suite

    def build_tests(self, test_class, path, workdir, module, expect_errors, tags):
        warning_errors = 'werror' in tags['tag']
        expect_warnings = 'warnings' in tags['tag']

        if expect_errors:
            if skip_c(tags) and 'cpp' in self.languages:
                languages = ['cpp']
            else:
                languages = self.languages[:1]
        else:
            languages = self.languages

        if 'c' in languages and skip_c(tags):
            languages = list(languages)
            languages.remove('c')
        if 'cpp' in languages and 'no-cpp' in tags['tag']:
            languages = list(languages)
            languages.remove('cpp')
        if not languages:
            return []

        language_levels = [2, 3] if 'all_language_levels' in tags['tag'] else [None]

        pythran_dir = self.pythran_dir
        if 'pythran' in tags['tag'] and not pythran_dir and 'cpp' in languages:
            import pythran.config
            try:
                pythran_ext = pythran.config.make_extension(python=True)
            except TypeError:  # old pythran version syntax
                pythran_ext = pythran.config.make_extension()
            pythran_dir = pythran_ext['include_dirs'][0]

        preparse_list = tags.get('preparse', ['id'])
        tests = [ self.build_test(test_class, path, workdir, module, tags, language, language_level,
                                  expect_errors, expect_warnings, warning_errors, preparse,
                                  pythran_dir if language == "cpp" else None)
                  for language in languages
                  for preparse in preparse_list
                  for language_level in language_levels
        ]
        return tests

    def build_test(self, test_class, path, workdir, module, tags, language, language_level,
                   expect_errors, expect_warnings, warning_errors, preparse, pythran_dir):
        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)
        if preparse != 'id':
            workdir += '_%s' % (preparse,)
        if language_level:
            workdir += '_cy%d' % (language_level,)
        return test_class(path, workdir, module, tags,
                          language=language,
                          preparse=preparse,
                          expect_errors=expect_errors,
                          expect_warnings=expect_warnings,
                          annotate=self.annotate,
                          cleanup_workdir=self.cleanup_workdir,
                          cleanup_sharedlibs=self.cleanup_sharedlibs,
                          cleanup_failures=self.cleanup_failures,
                          cython_only=self.cython_only,
                          doctest_selector=self.doctest_selector,
                          fork=self.fork,
                          language_level=language_level or self.language_level,
                          warning_errors=warning_errors,
                          test_determinism=self.test_determinism,
                          common_utility_dir=self.common_utility_dir,
                          pythran_dir=pythran_dir,
                          stats=self.stats)


def skip_c(tags):
    if 'cpp' in tags['tag']:
        return True

    # We don't want to create a distutils key in the
    # dictionary so we check before looping.
    if 'distutils' in tags:
        for option in tags['distutils']:
            splitted = option.split('=')
            if len(splitted) == 2:
                argument, value = splitted
                if argument.strip() == 'language' and value.strip() == 'c++':
                    return True
    return False


def filter_stderr(stderr_bytes):
    """
    Filter annoying warnings from output.
    """
    if b"Command line warning D9025" in stderr_bytes:
        # MSCV: cl : Command line warning D9025 : overriding '/Ox' with '/Od'
        stderr_bytes = b'\n'.join(
            line for line in stderr_bytes.splitlines()
            if b"Command line warning D9025" not in line)
    return stderr_bytes


class CythonCompileTestCase(unittest.TestCase):
    def __init__(self, test_directory, workdir, module, tags, language='c', preparse='id',
                 expect_errors=False, expect_warnings=False, annotate=False, cleanup_workdir=True,
                 cleanup_sharedlibs=True, cleanup_failures=True, cython_only=False, doctest_selector=None,
                 fork=True, language_level=2, warning_errors=False,
                 test_determinism=False,
                 common_utility_dir=None, pythran_dir=None, stats=None):
        self.test_directory = test_directory
        self.tags = tags
        self.workdir = workdir
        self.module = module
        self.language = language
        self.preparse = preparse
        self.name = module if self.preparse == "id" else "%s_%s" % (module, preparse)
        self.expect_errors = expect_errors
        self.expect_warnings = expect_warnings
        self.annotate = annotate
        self.cleanup_workdir = cleanup_workdir
        self.cleanup_sharedlibs = cleanup_sharedlibs
        self.cleanup_failures = cleanup_failures
        self.cython_only = cython_only
        self.doctest_selector = doctest_selector
        self.fork = fork
        self.language_level = language_level
        self.warning_errors = warning_errors
        self.test_determinism = test_determinism
        self.common_utility_dir = common_utility_dir
        self.pythran_dir = pythran_dir
        self.stats = stats
        unittest.TestCase.__init__(self)

    def shortDescription(self):
        return "compiling (%s%s%s) %s" % (
            self.language,
            "/cy2" if self.language_level == 2 else "/cy3" if self.language_level == 3 else "",
            "/pythran" if self.pythran_dir is not None else "",
            self.description_name()
        )

    def description_name(self):
        return self.name

    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',
                # 'cache_builtins',  # not currently supported due to incorrect global caching
            )
        ]
        self._saved_default_directives = list(Options.get_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
        is_cygwin = sys.platform == 'cygwin'

        if os.path.exists(self.workdir):
            if cleanup_c_files and cleanup_lib_files and not is_cygwin:
                shutil.rmtree(self.workdir, ignore_errors=True)
            else:
                for rmfile in os.listdir(self.workdir):
                    ext = os.path.splitext(rmfile)[1]
                    if not cleanup_c_files:
                        # Keep C, C++ files, header files, preprocessed sources
                        # and assembly sources (typically the .i and .s files
                        # are intentionally generated when -save-temps is given)
                        if ext in (".c", ".cpp", ".h", ".i", ".ii", ".s"):
                            continue
                        if ext == ".html" and rmfile.startswith(self.module):
                            continue

                    is_shared_obj = ext in (".so", ".dll")

                    if not cleanup_lib_files and is_shared_obj:
                        continue

                    try:
                        rmfile = os.path.join(self.workdir, rmfile)
                        if os.path.isdir(rmfile):
                            shutil.rmtree(rmfile, ignore_errors=True)
                        elif is_cygwin and is_shared_obj:
                            # Delete later
                            _to_clean.append(rmfile)
                        else:
                            os.remove(rmfile)
                    except IOError:
                        pass

                if cleanup_c_files and cleanup_lib_files and is_cygwin:
                    # Finally, remove the work dir itself
                    _to_clean.append(self.workdir)

        if cleanup_c_files and os.path.exists(self.workdir + '-again'):
            shutil.rmtree(self.workdir + '-again', ignore_errors=True)


    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.expect_warnings, 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):
        if self.preparse and self.preparse != 'id':
            preparse_func = globals()[self.preparse]
            def copy(src, dest):
                with open(src) as fin:
                    with open(dest, 'w') as fout:
                        fout.write(preparse_func(fin.read()))
        else:
            # 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')

        from Cython.Utils import detect_opened_file_encoding
        with io_open(source_file, 'rb') as f:
            # encoding is passed to ErrorWriter but not used on the source
            # since it is sometimes deliberately wrong
            encoding = detect_opened_file_encoding(f, default=None)

        with io_open(source_file, 'r', encoding='ISO-8859-1') as source_and_output:
            error_writer = warnings_writer = None
            out = io_open(os.path.join(workdir, module + os.path.splitext(source_file)[1]),
                          'w', encoding='ISO-8859-1')
            try:
                for line in source_and_output:
                    if line.startswith("_ERRORS"):
                        out.close()
                        out = error_writer = ErrorWriter(encoding=encoding)
                    elif line.startswith("_WARNINGS"):
                        out.close()
                        out = warnings_writer = ErrorWriter(encoding=encoding)
                    else:
                        out.write(line)
            finally:
                out.close()

        return (error_writer.geterrors() if error_writer else [],
                warnings_writer.geterrors() if warnings_writer else [])

    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)

        if self.preparse == 'id':
            source = self.find_module_source_file(
                os.path.join(test_directory, module + '.pyx'))
        else:
            self.copy_files(test_directory, targetdir, [module + '.pyx'])
            source = os.path.join(targetdir, module + '.pyx')
        target = os.path.join(targetdir, self.build_target_filename(module))

        if extra_compile_options is None:
            extra_compile_options = {}

        if 'allow_unknown_names' in self.tags['tag']:
            from Cython.Compiler import Options
            Options.error_on_unknown_names = False

        try:
            CompilationOptions
        except NameError:
            from Cython.Compiler.Options import CompilationOptions
            from Cython.Compiler.Main import compile as cython_compile
            from Cython.Compiler.Options import default_options
        common_utility_include_dir = self.common_utility_dir

        options = CompilationOptions(
            default_options,
            include_path = include_dirs,
            output_file = target,
            annotate = annotate,
            use_listing_file = False,
            cplus = self.language == 'cpp',
            np_pythran = self.pythran_dir is not None,
            language_level = self.language_level,
            generate_pxi = False,
            evaluate_tree_assertions = True,
            common_utility_include_dir = common_utility_include_dir,
            **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(get_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[:]

            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)