Python code coverage for Lib/test/test_importlib/test_locks.py

#	count	content
1	n/a	from . import util as test_util
2	n/a
3	n/a	init = test_util.import_importlib('importlib')
4	n/a
5	n/a	import sys
6	n/a	import unittest
7	n/a	import weakref
8	n/a
9	n/a	from test import support
10	n/a
11	n/a	try:
12	n/a	import threading
13	n/a	except ImportError:
14	n/a	threading = None
15	n/a	else:
16	n/a	from test import lock_tests
17	n/a
18	n/a	if threading is not None:
19	n/a	class ModuleLockAsRLockTests:
20	n/a	locktype = classmethod(lambda cls: cls.LockType("some_lock"))
21	n/a
22	n/a	# _is_owned() unsupported
23	n/a	test__is_owned = None
24	n/a	# acquire(blocking=False) unsupported
25	n/a	test_try_acquire = None
26	n/a	test_try_acquire_contended = None
27	n/a	# `with` unsupported
28	n/a	test_with = None
29	n/a	# acquire(timeout=...) unsupported
30	n/a	test_timeout = None
31	n/a	# _release_save() unsupported
32	n/a	test_release_save_unacquired = None
33	n/a	# lock status in repr unsupported
34	n/a	test_repr = None
35	n/a	test_locked_repr = None
36	n/a
37	n/a	LOCK_TYPES = {kind: splitinit._bootstrap._ModuleLock
38	n/a	for kind, splitinit in init.items()}
39	n/a
40	n/a	(Frozen_ModuleLockAsRLockTests,
41	n/a	Source_ModuleLockAsRLockTests
42	n/a	) = test_util.test_both(ModuleLockAsRLockTests, lock_tests.RLockTests,
43	n/a	LockType=LOCK_TYPES)
44	n/a	else:
45	n/a	LOCK_TYPES = {}
46	n/a
47	n/a	class Frozen_ModuleLockAsRLockTests(unittest.TestCase):
48	n/a	pass
49	n/a
50	n/a	class Source_ModuleLockAsRLockTests(unittest.TestCase):
51	n/a	pass
52	n/a
53	n/a
54	n/a	if threading is not None:
55	n/a	class DeadlockAvoidanceTests:
56	n/a
57	n/a	def setUp(self):
58	n/a	try:
59	n/a	self.old_switchinterval = sys.getswitchinterval()
60	n/a	support.setswitchinterval(0.000001)
61	n/a	except AttributeError:
62	n/a	self.old_switchinterval = None
63	n/a
64	n/a	def tearDown(self):
65	n/a	if self.old_switchinterval is not None:
66	n/a	sys.setswitchinterval(self.old_switchinterval)
67	n/a
68	n/a	def run_deadlock_avoidance_test(self, create_deadlock):
69	n/a	NLOCKS = 10
70	n/a	locks = [self.LockType(str(i)) for i in range(NLOCKS)]
71	n/a	pairs = [(locks[i], locks[(i+1)%NLOCKS]) for i in range(NLOCKS)]
72	n/a	if create_deadlock:
73	n/a	NTHREADS = NLOCKS
74	n/a	else:
75	n/a	NTHREADS = NLOCKS - 1
76	n/a	barrier = threading.Barrier(NTHREADS)
77	n/a	results = []
78	n/a
79	n/a	def _acquire(lock):
80	n/a	"""Try to acquire the lock. Return True on success,
81	n/a	False on deadlock."""
82	n/a	try:
83	n/a	lock.acquire()
84	n/a	except self.DeadlockError:
85	n/a	return False
86	n/a	else:
87	n/a	return True
88	n/a
89	n/a	def f():
90	n/a	a, b = pairs.pop()
91	n/a	ra = _acquire(a)
92	n/a	barrier.wait()
93	n/a	rb = _acquire(b)
94	n/a	results.append((ra, rb))
95	n/a	if rb:
96	n/a	b.release()
97	n/a	if ra:
98	n/a	a.release()
99	n/a	lock_tests.Bunch(f, NTHREADS).wait_for_finished()
100	n/a	self.assertEqual(len(results), NTHREADS)
101	n/a	return results
102	n/a
103	n/a	def test_deadlock(self):
104	n/a	results = self.run_deadlock_avoidance_test(True)
105	n/a	# At least one of the threads detected a potential deadlock on its
106	n/a	# second acquire() call. It may be several of them, because the
107	n/a	# deadlock avoidance mechanism is conservative.
108	n/a	nb_deadlocks = results.count((True, False))
109	n/a	self.assertGreaterEqual(nb_deadlocks, 1)
110	n/a	self.assertEqual(results.count((True, True)), len(results) - nb_deadlocks)
111	n/a
112	n/a	def test_no_deadlock(self):
113	n/a	results = self.run_deadlock_avoidance_test(False)
114	n/a	self.assertEqual(results.count((True, False)), 0)
115	n/a	self.assertEqual(results.count((True, True)), len(results))
116	n/a
117	n/a
118	n/a	DEADLOCK_ERRORS = {kind: splitinit._bootstrap._DeadlockError
119	n/a	for kind, splitinit in init.items()}
120	n/a
121	n/a	(Frozen_DeadlockAvoidanceTests,
122	n/a	Source_DeadlockAvoidanceTests
123	n/a	) = test_util.test_both(DeadlockAvoidanceTests,
124	n/a	LockType=LOCK_TYPES,
125	n/a	DeadlockError=DEADLOCK_ERRORS)
126	n/a	else:
127	n/a	DEADLOCK_ERRORS = {}
128	n/a
129	n/a	class Frozen_DeadlockAvoidanceTests(unittest.TestCase):
130	n/a	pass
131	n/a
132	n/a	class Source_DeadlockAvoidanceTests(unittest.TestCase):
133	n/a	pass
134	n/a
135	n/a
136	n/a	class LifetimeTests:
137	n/a
138	n/a	@property
139	n/a	def bootstrap(self):
140	n/a	return self.init._bootstrap
141	n/a
142	n/a	def test_lock_lifetime(self):
143	n/a	name = "xyzzy"
144	n/a	self.assertNotIn(name, self.bootstrap._module_locks)
145	n/a	lock = self.bootstrap._get_module_lock(name)
146	n/a	self.assertIn(name, self.bootstrap._module_locks)
147	n/a	wr = weakref.ref(lock)
148	n/a	del lock
149	n/a	support.gc_collect()
150	n/a	self.assertNotIn(name, self.bootstrap._module_locks)
151	n/a	self.assertIsNone(wr())
152	n/a
153	n/a	def test_all_locks(self):
154	n/a	support.gc_collect()
155	n/a	self.assertEqual(0, len(self.bootstrap._module_locks),
156	n/a	self.bootstrap._module_locks)
157	n/a
158	n/a
159	n/a	(Frozen_LifetimeTests,
160	n/a	Source_LifetimeTests
161	n/a	) = test_util.test_both(LifetimeTests, init=init)
162	n/a
163	n/a
164	n/a	@support.reap_threads
165	n/a	def test_main():
166	n/a	support.run_unittest(Frozen_ModuleLockAsRLockTests,
167	n/a	Source_ModuleLockAsRLockTests,
168	n/a	Frozen_DeadlockAvoidanceTests,
169	n/a	Source_DeadlockAvoidanceTests,
170	n/a	Frozen_LifetimeTests,
171	n/a	Source_LifetimeTests)
172	n/a
173	n/a
174	n/a	if __name__ == '__main__':
175	n/a	test_main()