summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_importlib/test_locks.py
diff options
context:
space:
mode:
authorBenjamin Peterson <benjamin@python.org>2012-09-29 18:14:19 (GMT)
committerBenjamin Peterson <benjamin@python.org>2012-09-29 18:14:19 (GMT)
commit14fb44e1bab9db128770f1d91d244916a669e7c3 (patch)
tree278d9ccc8f819b05f8114c3154087c70b88ed69c /Lib/test/test_importlib/test_locks.py
parent1764c80925795b6f059e961c5a352c5ece5a7fff (diff)
parent99a247fd01c1cd780c0c3ee1116657627f1ee744 (diff)
downloadcpython-14fb44e1bab9db128770f1d91d244916a669e7c3.zip
cpython-14fb44e1bab9db128770f1d91d244916a669e7c3.tar.gz
cpython-14fb44e1bab9db128770f1d91d244916a669e7c3.tar.bz2
merge mostly from default
Diffstat (limited to 'Lib/test/test_importlib/test_locks.py')
-rw-r--r--Lib/test/test_importlib/test_locks.py22
1 files changed, 18 insertions, 4 deletions
diff --git a/Lib/test/test_importlib/test_locks.py b/Lib/test/test_importlib/test_locks.py
index d36b71e..c373b11 100644
--- a/Lib/test/test_importlib/test_locks.py
+++ b/Lib/test/test_importlib/test_locks.py
@@ -1,4 +1,5 @@
from importlib import _bootstrap
+import sys
import time
import unittest
import weakref
@@ -41,6 +42,17 @@ else:
@unittest.skipUnless(threading, "threads needed for this test")
class DeadlockAvoidanceTests(unittest.TestCase):
+ def setUp(self):
+ try:
+ self.old_switchinterval = sys.getswitchinterval()
+ sys.setswitchinterval(0.000001)
+ except AttributeError:
+ self.old_switchinterval = None
+
+ def tearDown(self):
+ if self.old_switchinterval is not None:
+ sys.setswitchinterval(self.old_switchinterval)
+
def run_deadlock_avoidance_test(self, create_deadlock):
NLOCKS = 10
locks = [LockType(str(i)) for i in range(NLOCKS)]
@@ -75,10 +87,12 @@ class DeadlockAvoidanceTests(unittest.TestCase):
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)
+ # At least one of the threads detected a potential deadlock on its
+ # second acquire() call. It may be several of them, because the
+ # deadlock avoidance mechanism is conservative.
+ nb_deadlocks = results.count((True, False))
+ self.assertGreaterEqual(nb_deadlocks, 1)
+ self.assertEqual(results.count((True, True)), len(results) - nb_deadlocks)
def test_no_deadlock(self):
results = self.run_deadlock_avoidance_test(False)