diff options
| author | Brett Cannon <brett@python.org> | 2012-07-20 18:48:53 (GMT) |
|---|---|---|
| committer | Brett Cannon <brett@python.org> | 2012-07-20 18:48:53 (GMT) |
| commit | 45a5e3afe52ed89f298242143c5f7e2bb992ac63 (patch) | |
| tree | 2255a23f991f1452335d865939f6c30d7b2d7d48 /Lib/importlib/test/test_locks.py | |
| parent | 4afc1c08d04a9c083b654178ea4f947563510836 (diff) | |
| download | cpython-45a5e3afe52ed89f298242143c5f7e2bb992ac63.zip cpython-45a5e3afe52ed89f298242143c5f7e2bb992ac63.tar.gz cpython-45a5e3afe52ed89f298242143c5f7e2bb992ac63.tar.bz2 | |
Issue #15168: Move importlb.test to test.test_importlib.
This should make the Linux distros happy as it is now easier to leave
importlib's tests out of their base Python distribution.
Diffstat (limited to 'Lib/importlib/test/test_locks.py')
| -rw-r--r-- | Lib/importlib/test/test_locks.py | 115 |
1 files changed, 0 insertions, 115 deletions
diff --git a/Lib/importlib/test/test_locks.py b/Lib/importlib/test/test_locks.py deleted file mode 100644 index d36b71e..0000000 --- a/Lib/importlib/test/test_locks.py +++ /dev/null @@ -1,115 +0,0 @@ -from importlib import _bootstrap -import time -import unittest -import weakref - -from test import support - -try: - import threading -except ImportError: - threading = None -else: - from test import lock_tests - - -LockType = _bootstrap._ModuleLock -DeadlockError = _bootstrap._DeadlockError - - -if threading is not None: - class ModuleLockAsRLockTests(lock_tests.RLockTests): - locktype = staticmethod(lambda: LockType("some_lock")) - - # _is_owned() unsupported - test__is_owned = None - # acquire(blocking=False) unsupported - test_try_acquire = None - test_try_acquire_contended = None - # `with` unsupported - test_with = None - # acquire(timeout=...) unsupported - test_timeout = None - # _release_save() unsupported - test_release_save_unacquired = None - -else: - class ModuleLockAsRLockTests(unittest.TestCase): - pass - - -@unittest.skipUnless(threading, "threads needed for this test") -class DeadlockAvoidanceTests(unittest.TestCase): - - def run_deadlock_avoidance_test(self, create_deadlock): - NLOCKS = 10 - locks = [LockType(str(i)) for i in range(NLOCKS)] - pairs = [(locks[i], locks[(i+1)%NLOCKS]) for i in range(NLOCKS)] - if create_deadlock: - NTHREADS = NLOCKS - else: - NTHREADS = NLOCKS - 1 - barrier = threading.Barrier(NTHREADS) - results = [] - def _acquire(lock): - """Try to acquire the lock. Return True on success, False on deadlock.""" - try: - lock.acquire() - except DeadlockError: - return False - else: - return True - def f(): - a, b = pairs.pop() - ra = _acquire(a) - barrier.wait() - rb = _acquire(b) - results.append((ra, rb)) - if rb: - b.release() - if ra: - a.release() - lock_tests.Bunch(f, NTHREADS).wait_for_finished() - self.assertEqual(len(results), NTHREADS) - return results - - def test_deadlock(self): - results = self.run_deadlock_avoidance_test(True) - # One of the threads detected a potential deadlock on its second - # acquire() call. - self.assertEqual(results.count((True, False)), 1) - self.assertEqual(results.count((True, True)), len(results) - 1) - - def test_no_deadlock(self): - results = self.run_deadlock_avoidance_test(False) - self.assertEqual(results.count((True, False)), 0) - self.assertEqual(results.count((True, True)), len(results)) - - -class LifetimeTests(unittest.TestCase): - - def test_lock_lifetime(self): - name = "xyzzy" - self.assertNotIn(name, _bootstrap._module_locks) - lock = _bootstrap._get_module_lock(name) - self.assertIn(name, _bootstrap._module_locks) - wr = weakref.ref(lock) - del lock - support.gc_collect() - self.assertNotIn(name, _bootstrap._module_locks) - self.assertIsNone(wr()) - - def test_all_locks(self): - support.gc_collect() - self.assertEqual(0, len(_bootstrap._module_locks), _bootstrap._module_locks) - - -@support.reap_threads -def test_main(): - support.run_unittest(ModuleLockAsRLockTests, - DeadlockAvoidanceTests, - LifetimeTests) - - -if __name__ == '__main__': - test_main() |
