summaryrefslogtreecommitdiffstats
path: root/Lib/importlib/test/test_locks.py
diff options
context:
space:
mode:
authorBrett Cannon <brett@python.org>2012-07-20 18:48:53 (GMT)
committerBrett Cannon <brett@python.org>2012-07-20 18:48:53 (GMT)
commit45a5e3afe52ed89f298242143c5f7e2bb992ac63 (patch)
tree2255a23f991f1452335d865939f6c30d7b2d7d48 /Lib/importlib/test/test_locks.py
parent4afc1c08d04a9c083b654178ea4f947563510836 (diff)
downloadcpython-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.py115
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()