#!/usr/bin/env python

import atexit

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

import zlib

try:

    import platform

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

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

except (ImportError, AttributeError):

    IS_CPYTHON = True

    IS_PYPY = False

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 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.command.build_ext import build_ext as _build_ext

from distutils import sysconfig

from distutils import ccompiler

_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 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)

    _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',

}

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 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 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 is not None:

        compiler_version = gcc_version.group(1)

        if float(compiler_version) > 4.8:

            ext.extra_compile_args.append("-std=c++11")

        return ext    

    return EXCLUDE_EXT

def get_gcc_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:

       cc = ccompiler.get_default_compiler()

    if not cc:

        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)

    return gcc_version

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:bytesformat':  exclude_extension_in_pyver((3, 3), (3, 4)),  # no %-bytes formatting

}

# 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,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',

                                          'run.set_literals',

                                          ]),

    # 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,3) : (operator.lt, lambda x: x in ['build.package_compilation',

                                          'run.yield_from_py33',

                                          ]),

    (3,4): (operator.lt, lambda x: x in ['run.py34_signature',

                                         ]),

    (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

                                         ]),

}

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', '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

    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):

        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 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,

                 test_determinism,

                 common_utility_dir, pythran_dir=None):

        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

        self.test_determinism = test_determinism

        self.common_utility_dir = common_utility_dir

        self.pythran_dir = pythran_dir

    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 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)))

        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 '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')

        pythran_dir = self.pythran_dir

        if 'pythran' in tags['tag'] and not pythran_dir and 'cpp' in languages:

            import pythran.config

            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,

                                  expect_errors, expect_warnings, warning_errors, preparse,

                                  pythran_dir if language == "cpp" else None)

                  for language in languages

                  for preparse in preparse_list ]

        return tests

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

                   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' % str(preparse)

        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,

                          fork=self.fork,

                          language_level=self.language_level,

                          warning_errors=warning_errors,

                          test_determinism=self.test_determinism,

                          common_utility_dir=self.common_utility_dir,

                          pythran_dir=pythran_dir)

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,

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

                 test_determinism=False,

                 common_utility_dir=None, pythran_dir=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.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

        unittest.TestCase.__init__(self)

    def shortDescription(self):

        return "compiling (%s%s) %s" % (self.language, "/pythran" if self.pythran_dir is not None else "", 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')

        ]

        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):

                    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

                    is_shared_obj = rmfile.endswith(".so") or rmfile.endswith(".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):

                open(dest, 'w').write(preparse_func(open(src).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')

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

        error_writer = warnings_writer = None

        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 = error_writer = ErrorWriter()

                elif line.startswith("_WARNINGS"):

                    out.close()

                    out = warnings_writer = ErrorWriter()

                else:

                    out.write(line)

        finally:

            source_and_output.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)

        source = self.find_module_source_file(

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

        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 = {}