import doctest
import operator
import tempfile
+import traceback
try:
from StringIO import StringIO
except ImportError:
except ImportError:
import pickle
+try:
+ import threading
+except ImportError: # No threads, no problems
+ threading = None
+
WITH_CYTHON = True
VER_DEP_MODULES = {
# tests are excluded if 'CurrentPythonVersion OP VersionTuple', i.e.
- # (2,4) : (operator.le, ...) excludes ... when PyVer <= 2.4.x
+ # (2,4) : (operator.lt, ...) excludes ... when PyVer < 2.4.x
+ (2,4) : (operator.lt, lambda x: x in ['run.extern_builtins_T258',
+ 'run.builtin_sorted'
+ ]),
(2,5) : (operator.lt, lambda x: x in ['run.any',
'run.all',
]),
- (2,4) : (operator.le, lambda x: x in ['run.extern_builtins_T258'
- ]),
(2,6) : (operator.lt, lambda x: x in ['run.print_function',
'run.cython3',
]),
class TestBuilder(object):
def __init__(self, rootdir, workdir, selectors, exclude_selectors, annotate,
cleanup_workdir, cleanup_sharedlibs, with_pyregr, cython_only,
- languages, test_bugs, fork):
+ languages, test_bugs, fork, language_level):
self.rootdir = rootdir
self.workdir = workdir
self.selectors = selectors
self.languages = languages
self.test_bugs = test_bugs
self.fork = fork
+ self.language_level = language_level
def build_suite(self):
suite = unittest.TestSuite()
for test in self.build_tests(test_class, path, workdir,
module, expect_errors):
suite.addTest(test)
+ if context == 'run' and filename.endswith('.py'):
+ # additionally test file in real Python
+ suite.addTest(PureDoctestTestCase(module, os.path.join(path, filename)))
return suite
def build_tests(self, test_class, path, workdir, module, expect_errors):
cleanup_workdir=self.cleanup_workdir,
cleanup_sharedlibs=self.cleanup_sharedlibs,
cython_only=self.cython_only,
- fork=self.fork)
+ fork=self.fork,
+ language_level=self.language_level)
class CythonCompileTestCase(unittest.TestCase):
def __init__(self, test_directory, workdir, module, language='c',
expect_errors=False, annotate=False, cleanup_workdir=True,
- cleanup_sharedlibs=True, cython_only=False, fork=True):
+ cleanup_sharedlibs=True, cython_only=False, fork=True,
+ language_level=2):
self.test_directory = test_directory
self.workdir = workdir
self.module = module
self.cleanup_sharedlibs = cleanup_sharedlibs
self.cython_only = cython_only
self.fork = fork
+ self.language_level = language_level
unittest.TestCase.__init__(self)
def shortDescription(self):
if is_related(filename) and os.path.isfile(os.path.join(workdir, filename)) ]
def split_source_and_output(self, test_directory, module, workdir):
- source_file = os.path.join(test_directory, module) + '.pyx'
- source_and_output = codecs.open(
- self.find_module_source_file(source_file), 'rU', 'ISO-8859-1')
- out = codecs.open(os.path.join(workdir, module + '.pyx'),
- 'w', 'ISO-8859-1')
- for line in source_and_output:
- last_line = line
- if line.startswith("_ERRORS"):
- out.close()
- out = ErrorWriter()
- else:
- out.write(line)
+ source_file = self.find_module_source_file(os.path.join(test_directory, module) + '.pyx')
+ source_and_output = codecs.open(source_file, 'rU', 'ISO-8859-1')
+ try:
+ out = codecs.open(os.path.join(workdir, module + os.path.splitext(source_file)[1]),
+ 'w', 'ISO-8859-1')
+ for line in source_and_output:
+ last_line = line
+ if line.startswith("_ERRORS"):
+ out.close()
+ out = ErrorWriter()
+ else:
+ out.write(line)
+ finally:
+ source_and_output.close()
try:
geterrors = out.geterrors
except AttributeError:
+ out.close()
return []
else:
return geterrors()
annotate = annotate,
use_listing_file = False,
cplus = self.language == 'cpp',
+ language_level = self.language_level,
generate_pxi = False,
evaluate_tree_assertions = True,
)
result.startTest(self)
try:
self.setUp()
- self.runCompileTest()
- if not self.cython_only:
- self.run_doctests(self.module, result)
+ try:
+ self.runCompileTest()
+ if not self.cython_only:
+ self.run_doctests(self.module, result)
+ finally:
+ check_thread_termination()
except Exception:
result.addError(self, sys.exc_info())
result.stopTest(self)
output = open(result_file, 'wb')
pickle.dump(partial_result.data(), output)
except:
- import traceback
traceback.print_exc()
finally:
try: output.close()
try: os.unlink(result_file)
except: pass
+class PureDoctestTestCase(unittest.TestCase):
+ def __init__(self, module_name, module_path):
+ self.module_name = module_name
+ self.module_path = module_path
+ unittest.TestCase.__init__(self, 'run')
+
+ def shortDescription(self):
+ return "running pure doctests in %s" % self.module_name
+
+ def run(self, result=None):
+ if result is None:
+ result = self.defaultTestResult()
+ loaded_module_name = 'pure_doctest__' + self.module_name
+ result.startTest(self)
+ try:
+ self.setUp()
+
+ import imp
+ m = imp.load_source(loaded_module_name, self.module_path)
+ try:
+ doctest.DocTestSuite(m).run(result)
+ finally:
+ del m
+ if loaded_module_name in sys.modules:
+ del sys.modules[loaded_module_name]
+ check_thread_termination()
+ except Exception:
+ result.addError(self, sys.exc_info())
+ result.stopTest(self)
+ try:
+ self.tearDown()
+ except Exception:
+ pass
is_private_field = re.compile('^_[^_]').match
result.startTest(self)
try:
self.setUp()
- self.runCompileTest()
- unittest.defaultTestLoader.loadTestsFromName(self.module).run(result)
+ try:
+ self.runCompileTest()
+ unittest.defaultTestLoader.loadTestsFromName(self.module).run(result)
+ finally:
+ check_thread_termination()
except Exception:
result.addError(self, sys.exc_info())
result.stopTest(self)
for f in filenames:
if file_matches(f):
if not f.endswith('.py'): continue
- filepath = os.path.join(dirpath, f)[:-len(".py")]
+ filepath = os.path.join(dirpath, f)
+ if os.path.getsize(filepath) == 0: continue
+ if 'no doctest' in open(filepath).next(): continue
+ filepath = filepath[:-len(".py")]
modulename = module_prefix + filepath[len(path)+1:].replace(os.path.sep, '.')
if not [ 1 for match in selectors if match(modulename) ]:
continue
def __init__(self, list_file):
self.excludes = {}
- for line in open(list_file).readlines():
- line = line.strip()
- if line and line[0] != '#':
- self.excludes[line.split()[0]] = True
+ f = open(list_file)
+ try:
+ for line in f.readlines():
+ line = line.strip()
+ if line and line[0] != '#':
+ self.excludes[line.split()[0]] = True
+ finally:
+ f.close()
def __call__(self, testname):
return testname in self.excludes or testname.split('.')[-1] in self.excludes
''')
sys.path.insert(0, cy3_dir)
+class PendingThreadsError(RuntimeError):
+ pass
-if __name__ == '__main__':
+threads_seen = []
+
+def check_thread_termination(ignore_seen=True):
+ if threading is None: # no threading enabled in CPython
+ return
+ current = threading.currentThread()
+ blocking_threads = []
+ for t in threading.enumerate():
+ if not t.isAlive() or t == current:
+ continue
+ t.join(timeout=2)
+ if t.isAlive():
+ if not ignore_seen:
+ blocking_threads.append(t)
+ continue
+ for seen in threads_seen:
+ if t is seen:
+ break
+ else:
+ threads_seen.append(t)
+ blocking_threads.append(t)
+ if not blocking_threads:
+ return
+ sys.stderr.write("warning: left-over threads found after running test:\n")
+ for t in blocking_threads:
+ sys.stderr.write('...%s\n' % repr(t))
+ raise PendingThreadsError("left-over threads found after running test")
+
+def main():
from optparse import OptionParser
parser = OptionParser()
parser.add_option("--no-cleanup", dest="cleanup_workdir",
coverage.start()
if WITH_CYTHON:
+ global CompilationOptions, pyrex_default_options, cython_compile
from Cython.Compiler.Main import \
CompilationOptions, \
default_options as pyrex_default_options, \
if WITH_CYTHON and options.language_level == 3:
sys.stderr.write("Using Cython language level 3.\n")
- from Cython.Compiler import Options
- Options.directive_defaults['language_level'] = 3
sys.stderr.write("\n")
options.annotate_source, options.cleanup_workdir,
options.cleanup_sharedlibs, options.pyregr,
options.cython_only, languages, test_bugs,
- options.fork)
+ options.fork, options.language_level)
test_suite.addTest(filetests.build_suite())
if options.system_pyregr and languages:
options.annotate_source, options.cleanup_workdir,
options.cleanup_sharedlibs, True,
options.cython_only, languages, test_bugs,
- options.fork)
+ options.fork, options.language_level)
test_suite.addTest(
filetests.handle_directory(
os.path.join(sys.prefix, 'lib', 'python'+sys.version[:3], 'test'),
import refnanny
sys.stderr.write("\n".join([repr(x) for x in refnanny.reflog]))
+ print("ALL DONE")
+
if options.exit_ok:
- sys.exit(0)
+ return_code = 0
else:
- sys.exit(not result.wasSuccessful())
+ return_code = not result.wasSuccessful()
+
+ try:
+ check_thread_termination(ignore_seen=False)
+ sys.exit(return_code)
+ except PendingThreadsError:
+ # normal program exit won't kill the threads, do it the hard way here
+ os._exit(return_code)
+
+if __name__ == '__main__':
+ try:
+ main()
+ except SystemExit: # <= Py2.4 ...
+ raise
+ except Exception:
+ traceback.print_exc()
+ try:
+ check_thread_termination(ignore_seen=False)
+ except PendingThreadsError:
+ # normal program exit won't kill the threads, do it the hard way here
+ os._exit(1)
projects / cython.git / blobdiff