#!/usr/bin/env python3

import atexit

import base64

import doctest

import gc

import glob

import heapq

import locale

import math

import operator

import os

import platform

import re

import shutil

import subprocess

import sys

import sysconfig

import time

import traceback

import unittest

import warnings

import zlib

from collections import defaultdict

from contextlib import contextmanager

from functools import partial

from tempfile import TemporaryDirectory

try:

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

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

IS_GRAAL = platform.python_implementation() == 'GraalVM'

except (ImportError, AttributeError):

IS_CPYTHON = True

IS_PYPY = False

IS_GRAAL = False

CAN_SYMLINK = sys.platform != 'win32' and hasattr(os, 'symlink')

SHARED_UTILITY_MODULE_NAME = '_cython_shared'

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

WITH_COMPILE = True

try:

# Py3.12+ doesn't have distutils any more and requires setuptools to provide it.

import setuptools

except ImportError:

pass

from distutils.command.build_ext import build_ext as _build_ext

_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

def import_refnanny():

try:

# try test copy first

import refnanny

return refnanny

except ImportError:

pass

import Cython.Runtime.refnanny

return Cython.Runtime.refnanny

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

}

if sys.version_info < (3, 14):

EXT_DEP_MODULES['tag:subinterpreters'] = 'interpreters_backport'

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_test_in_pyver(*versions):

return sys.version_info[:2] in versions

def exclude_test_on_platform(*platforms):

return sys.platform in platforms

def exclude_test_on_dev():

return sys.version_info.releaselevel != 'final'

include_debugger = IS_CPYTHON

def exclude_test_if_no_gdb(*, _has_gdb=[None]):

if not include_debugger:

_has_gdb[0] = False

if _has_gdb[0] is None:

try:

subprocess.check_call(["gdb", "--version"])

except (IOError, subprocess.CalledProcessError):

_has_gdb[0] = False

else:

_has_gdb[0] = True

return not _has_gdb[0]

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 as np

# Add paths for npyrandom and npymath libraries:

lib_path = [

os.path.abspath(os.path.join(np.get_include(), '..', '..', 'random', 'lib')),

os.path.abspath(os.path.join(np.get_include(), '..', 'lib'))

]

ext.library_dirs += lib_path

if sys.platform == "win32":

if platform.machine().lower() != 'arm64':

# For unknown reasons, Windows arm provides libnpymath.a instead of npymath.lib.

# See if we can get away without it.

ext.libraries += ["npymath"]

else:

ext.libraries += ["npymath", "m"]

ext.include_dirs.append(np.get_include())

if set_api17_macro and getattr(np, '__version__', '') not in ('1.19.0', '1.19.1'):

ext.define_macros.append(('NPY_NO_DEPRECATED_API', 'NPY_1_7_API_VERSION'))

del np

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_language_extension(language, std, min_gcc_version=None, min_clang_version=None, min_macos_version=None):

def _update_language_extension(ext):

# If the extension provides a -std=... option, and it's greater than the one

# we're about to give, assume that whatever C compiler we use will probably be ok with it.

extra_compile_args = []

already_has_std = False

if ext.extra_compile_args:

std_regex = re.compile(r"-std(?!lib).*?(?P<number>[0-9]+)")

for ca in ext.extra_compile_args:

match = std_regex.search(ca)

if match:

number = int(match.group("number"))

if number < std:

continue # and drop the argument

already_has_std = True

extra_compile_args.append(ca)

ext.extra_compile_args = extra_compile_args

use_gcc = use_clang = already_has_std

# check for a usable gcc version

gcc_version = get_gcc_version(ext.language)

if gcc_version:

if not already_has_std:

compiler_version = gcc_version.group(1)

if not min_gcc_version or float(compiler_version) >= float(min_gcc_version):

use_gcc = True

ext.extra_compile_args.append(f"-std={language}{std}")

if use_gcc:

return ext

# check for a usable clang version

clang_version = get_clang_version(ext.language)

if clang_version:

if not already_has_std:

compiler_version = clang_version.group(1)

if not min_clang_version or float(compiler_version) >= float(min_clang_version):

use_clang = True

ext.extra_compile_args.append(f"-std={language}{std}")

if sys.platform == "darwin":

if language == "c++":

ext.extra_compile_args.append("-stdlib=libc++")

if min_macos_version is not None:

ext.extra_compile_args.append("-mmacosx-version-min=" + min_macos_version)

if use_clang:

return ext

# no usable C compiler found => exclude the extension

return EXCLUDE_EXT

return _update_language_extension

def update_c_extension(c_std, min_gcc_version=None, min_clang_version=None, min_macos_version=None):

return update_language_extension("c", c_std, min_gcc_version, min_clang_version, min_macos_version)

def update_cpp_extension(cpp_std, min_gcc_version=None, min_clang_version=None, min_macos_version=None):

return update_language_extension("c++", cpp_std, min_gcc_version, min_clang_version, min_macos_version)

update_cpp11_extension = update_cpp_extension(11, min_gcc_version="4.9", min_macos_version="10.7")

update_cpp17_extension = update_cpp_extension(17, min_gcc_version="5.0", min_macos_version="10.13")

update_cpp20_extension = update_cpp_extension(20, min_gcc_version="11.0", min_clang_version="13.0", min_macos_version="10.13")

update_c11_extension = update_c_extension(11, min_gcc_version="4.7", min_clang_version="3.3")

def require_gcc(version):

def check(ext):

gcc_version = get_gcc_version(ext.language)

if gcc_version:

if float(gcc_version.group(1)) >= float(version):

return ext

return EXCLUDE_EXT

return check

def get_cc_version(language):

"""

finds gcc version using Popen

"""

cc = ''

if language == 'cpp':

cc = os.environ.get('CXX') or sysconfig.get_config_var('CXX')

if not cc:

cc = os.environ.get('CC') or 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 as exc:

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

(language, os.strerror(exc.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:

compiler_version = [int(num) for num in compiler_version.split('.')]

if compiler_version >= [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:cpp17': update_cpp17_extension,

'tag:cpp20': update_cpp20_extension,

'tag:c11': update_c11_extension,

'tag:trace' : update_linetrace_extension,

'tag:cppexecpolicies': require_gcc("9.1"),

}

TAG_EXCLUDERS = sorted({

'no-macos': exclude_test_on_platform('darwin'),

'pstats': exclude_test_in_pyver((3,12)),

'coverage': exclude_test_in_pyver((3,12)) or exclude_test_on_dev(),

'monitoring': exclude_test_in_pyver((3,12)),

'gdb': exclude_test_if_no_gdb(),

'trace': not IS_CPYTHON or exclude_test_in_pyver((3,12)),

}.items())

def iterate_matcher_fixer_dict(matchers_and_fixers):

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

if isinstance(matcher, str):

# lazy init

del matchers_and_fixers[matcher]

matcher = string_selector(matcher)

matchers_and_fixers[matcher] = fixer

yield matcher, fixer

# TODO: use tags

VER_DEP_MODULES = {

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

# (3,12) : (operator.lt, ...) excludes ... when PyVer < 3.12.x

# FIXME: fix? delete?

(3,4,999): (operator.gt, lambda x: x in ['run.initial_file_path',

]),

(3,12): (operator.ge, lambda x: x in [

'run.py_unicode_strings', # Py_UNICODE was removed

'compile.pylong', # PyLongObject changed its structure

'run.longintrepr', # PyLongObject changed its structure

'run.line_trace', # test implementation broken by sys.monitoring

]),

}

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

CDEFS = []

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

get = f._cache.get

def func(*args):

res = 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:

if line[0] != '#':

# 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

continue

if line[0] != '#':

break

parsed = parse_tag(line)

if parsed is not None:

tag, values = parsed.groups()

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

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

continue

if tag == 'tags':

raise RuntimeError("test tags use the 'tag' directive, not 'tags' (%s)" % filepath)

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_module_from_file(module_name, file_path, execute=True):

import importlib.util

spec = importlib.util.spec_from_file_location(module_name, file_path)

m = importlib.util.module_from_spec(spec)

if execute:

sys.modules[module_name] = m

spec.loader.exec_module(m)

return m

def import_ext(module_name, file_path=None):

if file_path:

return import_module_from_file(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:

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|performance hint):)?(?:.*:)?\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 = {'error': [], 'warning': [], 'performance hint': []}

for line in s.splitlines():

match = self.match_error(line)

if match:

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

results[message_type or 'error'].append((int(line), int(column), message.strip()))

return [

["%d:%d: %s" % values for values in sorted(results[key])]

for key in ('error', 'warning', 'performance hint')

]

def geterrors(self):

return self._collect()[0]

def getall(self):

return self._collect()

def close(self):

pass # ignore, only to match file-like interface

class Heap:

def __init__(self, max_length=8):

assert max_length > 0, max_length

self._heap = []

self._max_length = max_length

self._push = heapq.heappush

self.push = self._push_growing

def _push_growing(self, item):

heapq.heappush(self._heap, item)

if len(self._heap) >= self._max_length:

self.push = partial(heapq.heappushpop, self._heap)

def push_all(self, items):

for item in (items._heap if isinstance(items, Heap) else items):

self.push(item)

def max(self):

return max(self._heap)

def __iter__(self):

return iter(sorted(self._heap, reverse=True))

class Stats(object):

can_count_module_sizes = (

sys.platform == "darwin" or sys.platform == 'linux'

and shutil.which('strip')

)

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(partial(Heap, top_n))

self.module_sizes = {'count': 0, 'total': 0, 'largest': Heap(top_n), 'smallest': Heap(top_n)}

def add_time(self, name, language, metric, t, count=1):

self.test_counts[metric] += count

self.test_times[metric] += t

# min-heap => pop smallest/shortest until longest times remain

self.top_tests[metric].push((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]

self.top_tests[metric].push_all(stats.top_tests[metric])

for heap in ('largest', 'smallest'):

self.module_sizes[heap].push_all(stats.module_sizes[heap])

self.module_sizes['total'] += stats.module_sizes['total']

self.module_sizes['count'] += stats.module_sizes['count']

def update_module_size(self, base_dir, test_module_dll):

if not self.can_count_module_sizes:

return

stripped_path = test_module_dll + ".stripped"

subprocess.run(['strip', '-S', test_module_dll, "-o" , stripped_path])

if os.path.isfile(stripped_path):

module_size = os.path.getsize(stripped_path)

os.unlink(stripped_path)

relative_module_path = os.path.relpath(test_module_dll, os.path.dirname(base_dir))

self.module_sizes['largest'].push((module_size, relative_module_path))

self.module_sizes['smallest'].push((-module_size, relative_module_path))

self.module_sizes['total'] += module_size

self.module_sizes['count'] += 1

def update_module_sizes(self, test_dir):

if not self.can_count_module_sizes:

return

for module in glob.glob(os.path.join(test_dir, "**", "*.so")):

self.update_module_size(test_dir, module)

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(), key=operator.itemgetter(1), reverse=True):

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

if self.module_sizes['count']:

lines.append('Test module binary sizes (stripped):\n')

avg_size = self.module_sizes['total'] / self.module_sizes['count']

lines.append(f"Total: {self.module_sizes['total']:_d} bytes / {self.module_sizes['count']:_d} modules = {avg_size:_.2f} bytes/module\n")

modules = ', '.join([f"{name} ({-neg_size:_d})" for neg_size, name in self.module_sizes['smallest']])

lines.append(f"Smallest modules: {-self.module_sizes['smallest'].max()[0]:12_d} bytes - {modules}\n")

modules = ', '.join([f"{name} ({size:_d})" for size, name in self.module_sizes['largest']])

lines.append(f"Largest modules: {self.module_sizes['largest'].max()[0]:12_d} bytes - {modules}\n")

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, add_cython_import=False,

add_cpp_locals_extra_tests=False):

self.rootdir = rootdir

self.workdir = workdir

self.selectors = selectors

self.exclude_selectors = exclude_selectors

self.shard_num = options.shard_num

self.annotate = options.annotate_source

self.evaluate_tree_assertions = options.evaluate_tree_assertions

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.abi3audit = options.abi3audit

self.test_selector = re.compile(options.only_pattern).search if options.only_pattern else None

self.languages = languages

self.test_bugs = test_bugs

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.add_cython_import = add_cython_import

self.capture = options.capture

self.add_cpp_locals_extra_tests = add_cpp_locals_extra_tests

self.shared_utility = SHARED_UTILITY_MODULE_NAME if options.use_shared_utility else None

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', 'darwin'] and self.add_embedded_test and not IS_GRAAL):

# 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.TestLoader().loadTestsFromTestCase(EmbedTest))

return suite

def handle_directory(self, path, context):

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

os.makedirs(workdir, exist_ok=True)

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 self.cython_only or not WITH_COMPILE:

# EndToEnd tests always execute arbitrary build and test code

continue

if skip_limited(tags):

continue

if self.shared_utility is not None and 'shared_utility' not in tags['tag']:

# EndToEnd tests have their own build setup which doesn't use the shared module unless the tag says so.

continue

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, shard_num=self.shard_num))

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, filepath, mode == 'error', tags):

suite.addTest(test)

if mode == 'run' and ext == '.py' and not filename.startswith('test_') and not (

self.cython_only or self.shared_utility):

# 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, filepath, tags, stats=self.stats, shard_num=self.shard_num))

return suite

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

warning_errors = 'werror' in tags['tag']

expect_log = ("errors",) if expect_errors else ()

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

expect_log += ("warnings",)

if "perf_hints" in tags['tag']:

expect_log += ("perf_hints",)

extra_directives_list = [{}]

if expect_errors:

if skip_c(tags) and 'cpp' in self.languages:

languages = ['cpp']

else:

languages = self.languages[:1]

else:

languages = self.languages

if (self.add_cpp_locals_extra_tests and 'cpp' in languages and

'cpp' in tags['tag'] and not 'no-cpp-locals' in tags['tag']):

extra_directives_list.append({'cpp_locals': True})

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 []

if skip_limited(tags):

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]

add_cython_import = self.add_cython_import and module_path.endswith('.py')

preparse_list = tags.get('preparse', ['id'])

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

tags, language, language_level,

expect_log,

warning_errors, preparse,

pythran_dir if language == "cpp" else None,

add_cython_import=add_cython_import,

extra_directives=extra_directives)

for language in languages

for preparse in preparse_list

for language_level in language_levels

for extra_directives in extra_directives_list

]

return tests

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

expect_log, warning_errors, preparse, pythran_dir, add_cython_import,

extra_directives):

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

os.makedirs(language_workdir, exist_ok=True)

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

if preparse != 'id':

workdir += '_%s' % (preparse,)

if language_level:

workdir += '_cy%d' % (language_level,)

if extra_directives:

workdir += ('_directives_'+ '_'.join('%s_%s' % (k, v) for k,v in extra_directives.items()))

return test_class(path, workdir, module, module_path, tags,

language=language,

preparse=preparse,

expect_log=expect_log,

annotate=self.annotate,

cleanup_workdir=self.cleanup_workdir,

cleanup_sharedlibs=self.cleanup_sharedlibs,

cleanup_failures=self.cleanup_failures,

cython_only=self.cython_only,

test_selector=self.test_selector,

shard_num=self.shard_num,

language_level=language_level or self.language_level,

warning_errors=warning_errors,

evaluate_tree_assertions=self.evaluate_tree_assertions,

test_determinism=self.test_determinism,

common_utility_dir=self.common_utility_dir,

pythran_dir=pythran_dir,

stats=self.stats,

add_cython_import=add_cython_import,

extra_directives=extra_directives,

abi3audit=self.abi3audit,

shared_utility=self.shared_utility,

)