Python code coverage for Lib/test/test_dummy_thread.py

#	count	content
1	n/a	import _dummy_thread as _thread
2	n/a	import time
3	n/a	import queue
4	n/a	import random
5	n/a	import unittest
6	n/a	from test import support
7	n/a	from unittest import mock
8	n/a
9	n/a	DELAY = 0
10	n/a
11	n/a
12	n/a	class LockTests(unittest.TestCase):
13	n/a	"""Test lock objects."""
14	n/a
15	n/a	def setUp(self):
16	n/a	# Create a lock
17	n/a	self.lock = _thread.allocate_lock()
18	n/a
19	n/a	def test_initlock(self):
20	n/a	#Make sure locks start locked
21	n/a	self.assertFalse(self.lock.locked(),
22	n/a	"Lock object is not initialized unlocked.")
23	n/a
24	n/a	def test_release(self):
25	n/a	# Test self.lock.release()
26	n/a	self.lock.acquire()
27	n/a	self.lock.release()
28	n/a	self.assertFalse(self.lock.locked(),
29	n/a	"Lock object did not release properly.")
30	n/a
31	n/a	def test_LockType_context_manager(self):
32	n/a	with _thread.LockType():
33	n/a	pass
34	n/a	self.assertFalse(self.lock.locked(),
35	n/a	"Acquired Lock was not released")
36	n/a
37	n/a	def test_improper_release(self):
38	n/a	#Make sure release of an unlocked thread raises RuntimeError
39	n/a	self.assertRaises(RuntimeError, self.lock.release)
40	n/a
41	n/a	def test_cond_acquire_success(self):
42	n/a	#Make sure the conditional acquiring of the lock works.
43	n/a	self.assertTrue(self.lock.acquire(0),
44	n/a	"Conditional acquiring of the lock failed.")
45	n/a
46	n/a	def test_cond_acquire_fail(self):
47	n/a	#Test acquiring locked lock returns False
48	n/a	self.lock.acquire(0)
49	n/a	self.assertFalse(self.lock.acquire(0),
50	n/a	"Conditional acquiring of a locked lock incorrectly "
51	n/a	"succeeded.")
52	n/a
53	n/a	def test_uncond_acquire_success(self):
54	n/a	#Make sure unconditional acquiring of a lock works.
55	n/a	self.lock.acquire()
56	n/a	self.assertTrue(self.lock.locked(),
57	n/a	"Uncondional locking failed.")
58	n/a
59	n/a	def test_uncond_acquire_return_val(self):
60	n/a	#Make sure that an unconditional locking returns True.
61	n/a	self.assertIs(self.lock.acquire(1), True,
62	n/a	"Unconditional locking did not return True.")
63	n/a	self.assertIs(self.lock.acquire(), True)
64	n/a
65	n/a	def test_uncond_acquire_blocking(self):
66	n/a	#Make sure that unconditional acquiring of a locked lock blocks.
67	n/a	def delay_unlock(to_unlock, delay):
68	n/a	"""Hold on to lock for a set amount of time before unlocking."""
69	n/a	time.sleep(delay)
70	n/a	to_unlock.release()
71	n/a
72	n/a	self.lock.acquire()
73	n/a	start_time = int(time.time())
74	n/a	_thread.start_new_thread(delay_unlock,(self.lock, DELAY))
75	n/a	if support.verbose:
76	n/a	print()
77	n/a	print("*** Waiting for thread to release the lock "\
78	n/a	"(approx. %s sec.) ***" % DELAY)
79	n/a	self.lock.acquire()
80	n/a	end_time = int(time.time())
81	n/a	if support.verbose:
82	n/a	print("done")
83	n/a	self.assertGreaterEqual(end_time - start_time, DELAY,
84	n/a	"Blocking by unconditional acquiring failed.")
85	n/a
86	n/a	@mock.patch('time.sleep')
87	n/a	def test_acquire_timeout(self, mock_sleep):
88	n/a	"""Test invoking acquire() with a positive timeout when the lock is
89	n/a	already acquired. Ensure that time.sleep() is invoked with the given
90	n/a	timeout and that False is returned."""
91	n/a
92	n/a	self.lock.acquire()
93	n/a	retval = self.lock.acquire(waitflag=0, timeout=1)
94	n/a	self.assertTrue(mock_sleep.called)
95	n/a	mock_sleep.assert_called_once_with(1)
96	n/a	self.assertEqual(retval, False)
97	n/a
98	n/a	def test_lock_representation(self):
99	n/a	self.lock.acquire()
100	n/a	self.assertIn("locked", repr(self.lock))
101	n/a	self.lock.release()
102	n/a	self.assertIn("unlocked", repr(self.lock))
103	n/a
104	n/a
105	n/a	class MiscTests(unittest.TestCase):
106	n/a	"""Miscellaneous tests."""
107	n/a
108	n/a	def test_exit(self):
109	n/a	self.assertRaises(SystemExit, _thread.exit)
110	n/a
111	n/a	def test_ident(self):
112	n/a	self.assertIsInstance(_thread.get_ident(), int,
113	n/a	"_thread.get_ident() returned a non-integer")
114	n/a	self.assertNotEqual(_thread.get_ident(), 0,
115	n/a	"_thread.get_ident() returned 0")
116	n/a
117	n/a	def test_LockType(self):
118	n/a	self.assertIsInstance(_thread.allocate_lock(), _thread.LockType,
119	n/a	"_thread.LockType is not an instance of what "
120	n/a	"is returned by _thread.allocate_lock()")
121	n/a
122	n/a	def test_set_sentinel(self):
123	n/a	self.assertIsInstance(_thread._set_sentinel(), _thread.LockType,
124	n/a	"_thread._set_sentinel() did not return a "
125	n/a	"LockType instance.")
126	n/a
127	n/a	def test_interrupt_main(self):
128	n/a	#Calling start_new_thread with a function that executes interrupt_main
129	n/a	# should raise KeyboardInterrupt upon completion.
130	n/a	def call_interrupt():
131	n/a	_thread.interrupt_main()
132	n/a
133	n/a	self.assertRaises(KeyboardInterrupt,
134	n/a	_thread.start_new_thread,
135	n/a	call_interrupt,
136	n/a	tuple())
137	n/a
138	n/a	def test_interrupt_in_main(self):
139	n/a	self.assertRaises(KeyboardInterrupt, _thread.interrupt_main)
140	n/a
141	n/a	def test_stack_size_None(self):
142	n/a	retval = _thread.stack_size(None)
143	n/a	self.assertEqual(retval, 0)
144	n/a
145	n/a	def test_stack_size_not_None(self):
146	n/a	with self.assertRaises(_thread.error) as cm:
147	n/a	_thread.stack_size("")
148	n/a	self.assertEqual(cm.exception.args[0],
149	n/a	"setting thread stack size not supported")
150	n/a
151	n/a
152	n/a	class ThreadTests(unittest.TestCase):
153	n/a	"""Test thread creation."""
154	n/a
155	n/a	def test_arg_passing(self):
156	n/a	#Make sure that parameter passing works.
157	n/a	def arg_tester(queue, arg1=False, arg2=False):
158	n/a	"""Use to test _thread.start_new_thread() passes args properly."""
159	n/a	queue.put((arg1, arg2))
160	n/a
161	n/a	testing_queue = queue.Queue(1)
162	n/a	_thread.start_new_thread(arg_tester, (testing_queue, True, True))
163	n/a	result = testing_queue.get()
164	n/a	self.assertTrue(result[0] and result[1],
165	n/a	"Argument passing for thread creation "
166	n/a	"using tuple failed")
167	n/a
168	n/a	_thread.start_new_thread(
169	n/a	arg_tester,
170	n/a	tuple(),
171	n/a	{'queue':testing_queue, 'arg1':True, 'arg2':True})
172	n/a
173	n/a	result = testing_queue.get()
174	n/a	self.assertTrue(result[0] and result[1],
175	n/a	"Argument passing for thread creation "
176	n/a	"using kwargs failed")
177	n/a
178	n/a	_thread.start_new_thread(
179	n/a	arg_tester,
180	n/a	(testing_queue, True),
181	n/a	{'arg2':True})
182	n/a
183	n/a	result = testing_queue.get()
184	n/a	self.assertTrue(result[0] and result[1],
185	n/a	"Argument passing for thread creation using both tuple"
186	n/a	" and kwargs failed")
187	n/a
188	n/a	def test_multi_thread_creation(self):
189	n/a	def queue_mark(queue, delay):
190	n/a	time.sleep(delay)
191	n/a	queue.put(_thread.get_ident())
192	n/a
193	n/a	thread_count = 5
194	n/a	testing_queue = queue.Queue(thread_count)
195	n/a
196	n/a	if support.verbose:
197	n/a	print()
198	n/a	print("*** Testing multiple thread creation "
199	n/a	"(will take approx. %s to %s sec.) ***" % (
200	n/a	DELAY, thread_count))
201	n/a
202	n/a	for count in range(thread_count):
203	n/a	if DELAY:
204	n/a	local_delay = round(random.random(), 1)
205	n/a	else:
206	n/a	local_delay = 0
207	n/a	_thread.start_new_thread(queue_mark,
208	n/a	(testing_queue, local_delay))
209	n/a	time.sleep(DELAY)
210	n/a	if support.verbose:
211	n/a	print('done')
212	n/a	self.assertEqual(testing_queue.qsize(), thread_count,
213	n/a	"Not all %s threads executed properly "
214	n/a	"after %s sec." % (thread_count, DELAY))
215	n/a
216	n/a	def test_args_not_tuple(self):
217	n/a	"""
218	n/a	Test invoking start_new_thread() with a non-tuple value for "args".
219	n/a	Expect TypeError with a meaningful error message to be raised.
220	n/a	"""
221	n/a	with self.assertRaises(TypeError) as cm:
222	n/a	_thread.start_new_thread(mock.Mock(), [])
223	n/a	self.assertEqual(cm.exception.args[0], "2nd arg must be a tuple")
224	n/a
225	n/a	def test_kwargs_not_dict(self):
226	n/a	"""
227	n/a	Test invoking start_new_thread() with a non-dict value for "kwargs".
228	n/a	Expect TypeError with a meaningful error message to be raised.
229	n/a	"""
230	n/a	with self.assertRaises(TypeError) as cm:
231	n/a	_thread.start_new_thread(mock.Mock(), tuple(), kwargs=[])
232	n/a	self.assertEqual(cm.exception.args[0], "3rd arg must be a dict")
233	n/a
234	n/a	def test_SystemExit(self):
235	n/a	"""
236	n/a	Test invoking start_new_thread() with a function that raises
237	n/a	SystemExit.
238	n/a	The exception should be discarded.
239	n/a	"""
240	n/a	func = mock.Mock(side_effect=SystemExit())
241	n/a	try:
242	n/a	_thread.start_new_thread(func, tuple())
243	n/a	except SystemExit:
244	n/a	self.fail("start_new_thread raised SystemExit.")
245	n/a
246	n/a	@mock.patch('traceback.print_exc')
247	n/a	def test_RaiseException(self, mock_print_exc):
248	n/a	"""
249	n/a	Test invoking start_new_thread() with a function that raises exception.
250	n/a
251	n/a	The exception should be discarded and the traceback should be printed
252	n/a	via traceback.print_exc()
253	n/a	"""
254	n/a	func = mock.Mock(side_effect=Exception)
255	n/a	_thread.start_new_thread(func, tuple())
256	n/a	self.assertTrue(mock_print_exc.called)