Back to index

python3.2  3.2.2
Public Member Functions | Public Attributes
test.test_threaded_import.ThreadedImportTests Class Reference

List of all members.

Public Member Functions

def setUp
def tearDown
def check_parallel_module_init
def test_parallel_module_init
def test_parallel_meta_path
def test_parallel_path_hooks
def test_import_hangers
def test_circular_imports

Public Attributes

 old_random

Detailed Description

Definition at line 85 of file test_threaded_import.py.


Member Function Documentation

Definition at line 97 of file test_threaded_import.py.

00097 
00098     def check_parallel_module_init(self):
00099         if imp.lock_held():
00100             # This triggers on, e.g., from test import autotest.
00101             raise unittest.SkipTest("can't run when import lock is held")
00102 
00103         done = threading.Event()
00104         for N in (20, 50) * 3:
00105             if verbose:
00106                 print("Trying", N, "threads ...", end=' ')
00107             # Make sure that random and modulefinder get reimported freshly
00108             for modname in ['random', 'modulefinder']:
00109                 try:
00110                     del sys.modules[modname]
00111                 except KeyError:
00112                     pass
00113             errors = []
00114             done_tasks = []
00115             done.clear()
00116             for i in range(N):
00117                 thread.start_new_thread(task, (N, done, done_tasks, errors,))
00118             done.wait(60)
00119             self.assertFalse(errors)
00120             if verbose:
00121                 print("OK.")

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 87 of file test_threaded_import.py.

00087 
00088     def setUp(self):
00089         self.old_random = sys.modules.pop('random', None)

Here is the caller graph for this function:

Definition at line 90 of file test_threaded_import.py.

00090 
00091     def tearDown(self):
00092         # If the `random` module was already initialized, we restore the
00093         # old module at the end so that pickling tests don't fail.
00094         # See http://bugs.python.org/issue3657#msg110461
00095         if self.old_random is not None:
00096             sys.modules['random'] = self.old_random

Here is the caller graph for this function:

Definition at line 168 of file test_threaded_import.py.

00168 
00169     def test_circular_imports(self):
00170         # The goal of this test is to exercise implementations of the import
00171         # lock which use a per-module lock, rather than a global lock.
00172         # In these implementations, there is a possible deadlock with
00173         # circular imports, for example:
00174         # - thread 1 imports A (grabbing the lock for A) which imports B
00175         # - thread 2 imports B (grabbing the lock for B) which imports A
00176         # Such implementations should be able to detect such situations and
00177         # resolve them one way or the other, without freezing.
00178         # NOTE: our test constructs a slightly less trivial import cycle,
00179         # in order to better stress the deadlock avoidance mechanism.
00180         delay = 0.5
00181         os.mkdir(TESTFN)
00182         self.addCleanup(shutil.rmtree, TESTFN)
00183         sys.path.insert(0, TESTFN)
00184         self.addCleanup(sys.path.remove, TESTFN)
00185         for name, contents in circular_imports_modules.items():
00186             contents = contents % {'delay': delay}
00187             with open(os.path.join(TESTFN, name + ".py"), "wb") as f:
00188                 f.write(contents.encode('utf-8'))
00189             self.addCleanup(sys.modules.pop, name, None)
00190 
00191         results = []
00192         def import_ab():
00193             import A
00194             results.append(getattr(A, 'x', None))
00195         def import_ba():
00196             import B
00197             results.append(getattr(B, 'x', None))
00198         t1 = threading.Thread(target=import_ab)
00199         t2 = threading.Thread(target=import_ba)
00200         t1.start()
00201         t2.start()
00202         t1.join()
00203         t2.join()
00204         self.assertEqual(set(results), {'a', 'b'})
00205 

Here is the call graph for this function:

Definition at line 158 of file test_threaded_import.py.

00158 
00159     def test_import_hangers(self):
00160         # In case this test is run again, make sure the helper module
00161         # gets loaded from scratch again.
00162         try:
00163             del sys.modules['test.threaded_import_hangers']
00164         except KeyError:
00165             pass
00166         import test.threaded_import_hangers
00167         self.assertFalse(test.threaded_import_hangers.errors)

Here is the call graph for this function:

Definition at line 125 of file test_threaded_import.py.

00125 
00126     def test_parallel_meta_path(self):
00127         finder = Finder()
00128         sys.meta_path.append(finder)
00129         try:
00130             self.check_parallel_module_init()
00131             self.assertGreater(finder.numcalls, 0)
00132             self.assertEqual(finder.x, finder.numcalls)
00133         finally:
00134             sys.meta_path.remove(finder)

Here is the call graph for this function:

Definition at line 122 of file test_threaded_import.py.

Here is the call graph for this function:

Definition at line 135 of file test_threaded_import.py.

00135 
00136     def test_parallel_path_hooks(self):
00137         # Here the Finder instance is only used to check concurrent calls
00138         # to path_hook().
00139         finder = Finder()
00140         # In order for our path hook to be called at each import, we need
00141         # to flush the path_importer_cache, which we do by registering a
00142         # dedicated meta_path entry.
00143         flushing_finder = FlushingFinder()
00144         def path_hook(path):
00145             finder.find_module('')
00146             raise ImportError
00147         sys.path_hooks.append(path_hook)
00148         sys.meta_path.append(flushing_finder)
00149         try:
00150             # Flush the cache a first time
00151             flushing_finder.find_module('')
00152             numtests = self.check_parallel_module_init()
00153             self.assertGreater(finder.numcalls, 0)
00154             self.assertEqual(finder.x, finder.numcalls)
00155         finally:
00156             sys.meta_path.remove(flushing_finder)
00157             sys.path_hooks.remove(path_hook)

Here is the call graph for this function:


Member Data Documentation

Definition at line 88 of file test_threaded_import.py.


The documentation for this class was generated from the following file: