Python code coverage for Lib/dummy_thread.py

#	count	content
1	n/a	"""Drop-in replacement for the thread module.
2	n/a
3	n/a	Meant to be used as a brain-dead substitute so that threaded code does
4	n/a	not need to be rewritten for when the thread module is not present.
5	n/a
6	n/a	Suggested usage is::
7	n/a
8	n/a	try:
9	n/a	import thread
10	n/a	except ImportError:
11	n/a	import dummy_thread as thread
12	n/a
13	1	"""
14	n/a	# Exports only things specified by thread documentation;
15	n/a	# skipping obsolete synonyms allocate(), start_new(), exit_thread().
16	1	__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
17	1	'interrupt_main', 'LockType']
18	n/a
19	1	import traceback as _traceback
20	n/a
21	2	class error(Exception):
22	1	"""Dummy implementation of thread.error."""
23	n/a
24	1	def __init__(self, *args):
25	1	self.args = args
26	n/a
27	1	def start_new_thread(function, args, kwargs={}):
28	n/a	"""Dummy implementation of thread.start_new_thread().
29	n/a
30	n/a	Compatibility is maintained by making sure that ``args`` is a
31	n/a	tuple and ``kwargs`` is a dictionary. If an exception is raised
32	n/a	and it is SystemExit (which can be done by thread.exit()) it is
33	n/a	caught and nothing is done; all other exceptions are printed out
34	n/a	by using traceback.print_exc().
35	n/a
36	n/a	If the executed function calls interrupt_main the KeyboardInterrupt will be
37	n/a	raised when the function returns.
38	n/a
39	n/a	"""
40	20	if type(args) != type(tuple()):
41	0	raise TypeError("2nd arg must be a tuple")
42	20	if type(kwargs) != type(dict()):
43	0	raise TypeError("3rd arg must be a dict")
44	n/a	global _main
45	20	_main = False
46	20	try:
47	20	function(args, *kwargs)
48	0	except SystemExit:
49	0	pass
50	0	except:
51	0	_traceback.print_exc()
52	20	_main = True
53	n/a	global _interrupt
54	20	if _interrupt:
55	1	_interrupt = False
56	1	raise KeyboardInterrupt
57	n/a
58	1	def exit():
59	n/a	"""Dummy implementation of thread.exit()."""
60	1	raise SystemExit
61	n/a
62	1	def get_ident():
63	n/a	"""Dummy implementation of thread.get_ident().
64	n/a
65	n/a	Since this module should only be used when threadmodule is not
66	n/a	available, it is safe to assume that the current process is the
67	n/a	only thread. Thus a constant can be safely returned.
68	n/a	"""
69	109	return -1
70	n/a
71	1	def allocate_lock():
72	n/a	"""Dummy implementation of thread.allocate_lock()."""
73	54	return LockType()
74	n/a
75	1	def stack_size(size=None):
76	n/a	"""Dummy implementation of thread.stack_size()."""
77	0	if size is not None:
78	0	raise error("setting thread stack size not supported")
79	0	return 0
80	n/a
81	2	class LockType(object):
82	n/a	"""Class implementing dummy implementation of thread.LockType.
83	n/a
84	n/a	Compatibility is maintained by maintaining self.locked_status
85	n/a	which is a boolean that stores the state of the lock. Pickling of
86	n/a	the lock, though, should not be done since if the thread module is
87	n/a	then used with an unpickled ``lock()`` from here problems could
88	n/a	occur from this class not having atomic methods.
89	n/a
90	1	"""
91	n/a
92	1	def __init__(self):
93	54	self.locked_status = False
94	n/a
95	1	def acquire(self, waitflag=None):
96	n/a	"""Dummy implementation of acquire().
97	n/a
98	n/a	For blocking calls, self.locked_status is automatically set to
99	n/a	True and returned appropriately based on value of
100	n/a	``waitflag``. If it is non-blocking, then the value is
101	n/a	actually checked and not set if it is already acquired. This
102	n/a	is all done so that threading.Condition's assert statements
103	n/a	aren't triggered and throw a little fit.
104	n/a
105	n/a	"""
106	182	if waitflag is None or waitflag:
107	138	self.locked_status = True
108	138	return True
109	n/a	else:
110	44	if not self.locked_status:
111	2	self.locked_status = True
112	2	return True
113	n/a	else:
114	42	return False
115	n/a
116	1	__enter__ = acquire
117	n/a
118	1	def __exit__(self, typ, val, tb):
119	41	self.release()
120	n/a
121	1	def release(self):
122	n/a	"""Release the dummy lock."""
123	n/a	# XXX Perhaps shouldn't actually bother to test? Could lead
124	n/a	# to problems for complex, threaded code.
125	135	if not self.locked_status:
126	1	raise error
127	134	self.locked_status = False
128	134	return True
129	n/a
130	1	def locked(self):
131	3	return self.locked_status
132	n/a
133	n/a	# Used to signal that interrupt_main was called in a "thread"
134	1	_interrupt = False
135	n/a	# True when not executing in a "thread"
136	1	_main = True
137	n/a
138	1	def interrupt_main():
139	n/a	"""Set _interrupt flag to True to have start_new_thread raise
140	n/a	KeyboardInterrupt upon exiting."""
141	2	if _main:
142	1	raise KeyboardInterrupt
143	n/a	else:
144	n/a	global _interrupt
145	1	_interrupt = True