import functools
import importlib
from os.path import normcase
+try:
+ from concurrent.futures import ThreadPoolExecutor
+except ImportError:
+ ThreadPoolExecutor = None
from test.support import run_unittest, TESTFN, DirsOnSysPath
-from test.support import multiprocessing as has_multiprocessing
from test.script_helper import assert_python_ok, assert_python_failure
from test import inspect_fodder as mod
from test import inspect_fodder2 as mod2
self.assertEqual(lines[:-1], inspect.getsource(module).splitlines())
self.assertEqual(err, b'')
- @unittest.skipIf(not has_multiprocessing,
+ @unittest.skipIf(ThreadPoolExecutor is None,
'multiprocessing required to test __qualname__ for source files')
def test_qualname_source(self):
- module = importlib.import_module('concurrent.futures')
- member = getattr(module, 'ThreadPoolExecutor')
rc, out, err = assert_python_ok('-m', 'inspect',
'concurrent.futures:ThreadPoolExecutor')
lines = out.decode().splitlines()
# ignore the final newline
self.assertEqual(lines[:-1],
- inspect.getsource(member).splitlines())
+ inspect.getsource(ThreadPoolExecutor).splitlines())
self.assertEqual(err, b'')
def test_builtins(self):