Python code coverage for Lib/test/test_dummy_threading.py

#	count	content
1	n/a	from test import support
2	n/a	import unittest
3	n/a	import dummy_threading as _threading
4	n/a	import time
5	n/a
6	n/a	class DummyThreadingTestCase(unittest.TestCase):
7	n/a
8	n/a	class TestThread(_threading.Thread):
9	n/a
10	n/a	def run(self):
11	n/a	global running
12	n/a	global sema
13	n/a	global mutex
14	n/a	# Uncomment if testing another module, such as the real 'threading'
15	n/a	# module.
16	n/a	#delay = random.random() * 2
17	n/a	delay = 0
18	n/a	if support.verbose:
19	n/a	print('task', self.name, 'will run for', delay, 'sec')
20	n/a	sema.acquire()
21	n/a	mutex.acquire()
22	n/a	running += 1
23	n/a	if support.verbose:
24	n/a	print(running, 'tasks are running')
25	n/a	mutex.release()
26	n/a	time.sleep(delay)
27	n/a	if support.verbose:
28	n/a	print('task', self.name, 'done')
29	n/a	mutex.acquire()
30	n/a	running -= 1
31	n/a	if support.verbose:
32	n/a	print(self.name, 'is finished.', running, 'tasks are running')
33	n/a	mutex.release()
34	n/a	sema.release()
35	n/a
36	n/a	def setUp(self):
37	n/a	self.numtasks = 10
38	n/a	global sema
39	n/a	sema = _threading.BoundedSemaphore(value=3)
40	n/a	global mutex
41	n/a	mutex = _threading.RLock()
42	n/a	global running
43	n/a	running = 0
44	n/a	self.threads = []
45	n/a
46	n/a	def test_tasks(self):
47	n/a	for i in range(self.numtasks):
48	n/a	t = self.TestThread(name="<thread %d>"%i)
49	n/a	self.threads.append(t)
50	n/a	t.start()
51	n/a
52	n/a	if support.verbose:
53	n/a	print('waiting for all tasks to complete')
54	n/a	for t in self.threads:
55	n/a	t.join()
56	n/a	if support.verbose:
57	n/a	print('all tasks done')
58	n/a
59	n/a	if __name__ == '__main__':
60	n/a	unittest.main()