Python code coverage for Lib/test/test_threading_local.py

#	count	content
1	n/a	import unittest
2	n/a	from doctest import DocTestSuite
3	n/a	from test import support
4	n/a	import weakref
5	n/a	import gc
6	n/a
7	n/a	# Modules under test
8	n/a	_thread = support.import_module('_thread')
9	n/a	threading = support.import_module('threading')
10	n/a	import _threading_local
11	n/a
12	n/a
13	n/a	class Weak(object):
14	n/a	pass
15	n/a
16	n/a	def target(local, weaklist):
17	n/a	weak = Weak()
18	n/a	local.weak = weak
19	n/a	weaklist.append(weakref.ref(weak))
20	n/a
21	n/a
22	n/a	class BaseLocalTest:
23	n/a
24	n/a	def test_local_refs(self):
25	n/a	self._local_refs(20)
26	n/a	self._local_refs(50)
27	n/a	self._local_refs(100)
28	n/a
29	n/a	def _local_refs(self, n):
30	n/a	local = self._local()
31	n/a	weaklist = []
32	n/a	for i in range(n):
33	n/a	t = threading.Thread(target=target, args=(local, weaklist))
34	n/a	t.start()
35	n/a	t.join()
36	n/a	del t
37	n/a
38	n/a	gc.collect()
39	n/a	self.assertEqual(len(weaklist), n)
40	n/a
41	n/a	# XXX _threading_local keeps the local of the last stopped thread alive.
42	n/a	deadlist = [weak for weak in weaklist if weak() is None]
43	n/a	self.assertIn(len(deadlist), (n-1, n))
44	n/a
45	n/a	# Assignment to the same thread local frees it sometimes (!)
46	n/a	local.someothervar = None
47	n/a	gc.collect()
48	n/a	deadlist = [weak for weak in weaklist if weak() is None]
49	n/a	self.assertIn(len(deadlist), (n-1, n), (n, len(deadlist)))
50	n/a
51	n/a	def test_derived(self):
52	n/a	# Issue 3088: if there is a threads switch inside the __init__
53	n/a	# of a threading.local derived class, the per-thread dictionary
54	n/a	# is created but not correctly set on the object.
55	n/a	# The first member set may be bogus.
56	n/a	import time
57	n/a	class Local(self._local):
58	n/a	def __init__(self):
59	n/a	time.sleep(0.01)
60	n/a	local = Local()
61	n/a
62	n/a	def f(i):
63	n/a	local.x = i
64	n/a	# Simply check that the variable is correctly set
65	n/a	self.assertEqual(local.x, i)
66	n/a
67	n/a	with support.start_threads(threading.Thread(target=f, args=(i,))
68	n/a	for i in range(10)):
69	n/a	pass
70	n/a
71	n/a	def test_derived_cycle_dealloc(self):
72	n/a	# http://bugs.python.org/issue6990
73	n/a	class Local(self._local):
74	n/a	pass
75	n/a	locals = None
76	n/a	passed = False
77	n/a	e1 = threading.Event()
78	n/a	e2 = threading.Event()
79	n/a
80	n/a	def f():
81	n/a	nonlocal passed
82	n/a	# 1) Involve Local in a cycle
83	n/a	cycle = [Local()]
84	n/a	cycle.append(cycle)
85	n/a	cycle[0].foo = 'bar'
86	n/a
87	n/a	# 2) GC the cycle (triggers threadmodule.c::local_clear
88	n/a	# before local_dealloc)
89	n/a	del cycle
90	n/a	gc.collect()
91	n/a	e1.set()
92	n/a	e2.wait()
93	n/a
94	n/a	# 4) New Locals should be empty
95	n/a	passed = all(not hasattr(local, 'foo') for local in locals)
96	n/a
97	n/a	t = threading.Thread(target=f)
98	n/a	t.start()
99	n/a	e1.wait()
100	n/a
101	n/a	# 3) New Locals should recycle the original's address. Creating
102	n/a	# them in the thread overwrites the thread state and avoids the
103	n/a	# bug
104	n/a	locals = [Local() for i in range(10)]
105	n/a	e2.set()
106	n/a	t.join()
107	n/a
108	n/a	self.assertTrue(passed)
109	n/a
110	n/a	def test_arguments(self):
111	n/a	# Issue 1522237
112	n/a	class MyLocal(self._local):
113	n/a	def __init__(self, args, *kwargs):
114	n/a	pass
115	n/a
116	n/a	MyLocal(a=1)
117	n/a	MyLocal(1)
118	n/a	self.assertRaises(TypeError, self._local, a=1)
119	n/a	self.assertRaises(TypeError, self._local, 1)
120	n/a
121	n/a	def _test_one_class(self, c):
122	n/a	self._failed = "No error message set or cleared."
123	n/a	obj = c()
124	n/a	e1 = threading.Event()
125	n/a	e2 = threading.Event()
126	n/a
127	n/a	def f1():
128	n/a	obj.x = 'foo'
129	n/a	obj.y = 'bar'
130	n/a	del obj.y
131	n/a	e1.set()
132	n/a	e2.wait()
133	n/a
134	n/a	def f2():
135	n/a	try:
136	n/a	foo = obj.x
137	n/a	except AttributeError:
138	n/a	# This is expected -- we haven't set obj.x in this thread yet!
139	n/a	self._failed = "" # passed
140	n/a	else:
141	n/a	self._failed = ('Incorrectly got value %r from class %r\n' %
142	n/a	(foo, c))
143	n/a	sys.stderr.write(self._failed)
144	n/a
145	n/a	t1 = threading.Thread(target=f1)
146	n/a	t1.start()
147	n/a	e1.wait()
148	n/a	t2 = threading.Thread(target=f2)
149	n/a	t2.start()
150	n/a	t2.join()
151	n/a	# The test is done; just let t1 know it can exit, and wait for it.
152	n/a	e2.set()
153	n/a	t1.join()
154	n/a
155	n/a	self.assertFalse(self._failed, self._failed)
156	n/a
157	n/a	def test_threading_local(self):
158	n/a	self._test_one_class(self._local)
159	n/a
160	n/a	def test_threading_local_subclass(self):
161	n/a	class LocalSubclass(self._local):
162	n/a	"""To test that subclasses behave properly."""
163	n/a	self._test_one_class(LocalSubclass)
164	n/a
165	n/a	def _test_dict_attribute(self, cls):
166	n/a	obj = cls()
167	n/a	obj.x = 5
168	n/a	self.assertEqual(obj.__dict__, {'x': 5})
169	n/a	with self.assertRaises(AttributeError):
170	n/a	obj.__dict__ = {}
171	n/a	with self.assertRaises(AttributeError):
172	n/a	del obj.__dict__
173	n/a
174	n/a	def test_dict_attribute(self):
175	n/a	self._test_dict_attribute(self._local)
176	n/a
177	n/a	def test_dict_attribute_subclass(self):
178	n/a	class LocalSubclass(self._local):
179	n/a	"""To test that subclasses behave properly."""
180	n/a	self._test_dict_attribute(LocalSubclass)
181	n/a
182	n/a	def test_cycle_collection(self):
183	n/a	class X:
184	n/a	pass
185	n/a
186	n/a	x = X()
187	n/a	x.local = self._local()
188	n/a	x.local.x = x
189	n/a	wr = weakref.ref(x)
190	n/a	del x
191	n/a	gc.collect()
192	n/a	self.assertIsNone(wr())
193	n/a
194	n/a
195	n/a	class ThreadLocalTest(unittest.TestCase, BaseLocalTest):
196	n/a	_local = _thread._local
197	n/a
198	n/a	class PyThreadingLocalTest(unittest.TestCase, BaseLocalTest):
199	n/a	_local = _threading_local.local
200	n/a
201	n/a
202	n/a	def test_main():
203	n/a	suite = unittest.TestSuite()
204	n/a	suite.addTest(DocTestSuite('_threading_local'))
205	n/a	suite.addTest(unittest.makeSuite(ThreadLocalTest))
206	n/a	suite.addTest(unittest.makeSuite(PyThreadingLocalTest))
207	n/a
208	n/a	local_orig = _threading_local.local
209	n/a	def setUp(test):
210	n/a	_threading_local.local = _thread._local
211	n/a	def tearDown(test):
212	n/a	_threading_local.local = local_orig
213	n/a	suite.addTest(DocTestSuite('_threading_local',
214	n/a	setUp=setUp, tearDown=tearDown)
215	n/a	)
216	n/a
217	n/a	support.run_unittest(suite)
218	n/a
219	n/a	if __name__ == '__main__':
220	n/a	test_main()