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	n/a	"""
14	n/a	# Exports only things specified by thread documentation;
15	n/a	# skipping obsolete synonyms allocate(), start_new(), exit_thread().
16	n/a	__all__ = ['error', 'start_new_thread', 'exit', 'get_ident', 'allocate_lock',
17	n/a	'interrupt_main', 'LockType']
18	n/a
19	n/a	# A dummy value
20	n/a	TIMEOUT_MAX = 2**31
21	n/a
22	n/a	# NOTE: this module can be imported early in the extension building process,
23	n/a	# and so top level imports of other modules should be avoided. Instead, all
24	n/a	# imports are done when needed on a function-by-function basis. Since threads
25	n/a	# are disabled, the import lock should not be an issue anyway (??).
26	n/a
27	n/a	error = RuntimeError
28	n/a
29	n/a	def start_new_thread(function, args, kwargs={}):
30	n/a	"""Dummy implementation of _thread.start_new_thread().
31	n/a
32	n/a	Compatibility is maintained by making sure that ``args`` is a
33	n/a	tuple and ``kwargs`` is a dictionary. If an exception is raised
34	n/a	and it is SystemExit (which can be done by _thread.exit()) it is
35	n/a	caught and nothing is done; all other exceptions are printed out
36	n/a	by using traceback.print_exc().
37	n/a
38	n/a	If the executed function calls interrupt_main the KeyboardInterrupt will be
39	n/a	raised when the function returns.
40	n/a
41	n/a	"""
42	n/a	if type(args) != type(tuple()):
43	n/a	raise TypeError("2nd arg must be a tuple")
44	n/a	if type(kwargs) != type(dict()):
45	n/a	raise TypeError("3rd arg must be a dict")
46	n/a	global _main
47	n/a	_main = False
48	n/a	try:
49	n/a	function(args, *kwargs)
50	n/a	except SystemExit:
51	n/a	pass
52	n/a	except:
53	n/a	import traceback
54	n/a	traceback.print_exc()
55	n/a	_main = True
56	n/a	global _interrupt
57	n/a	if _interrupt:
58	n/a	_interrupt = False
59	n/a	raise KeyboardInterrupt
60	n/a
61	n/a	def exit():
62	n/a	"""Dummy implementation of _thread.exit()."""
63	n/a	raise SystemExit
64	n/a
65	n/a	def get_ident():
66	n/a	"""Dummy implementation of _thread.get_ident().
67	n/a
68	n/a	Since this module should only be used when _threadmodule is not
69	n/a	available, it is safe to assume that the current process is the
70	n/a	only thread. Thus a constant can be safely returned.
71	n/a	"""
72	n/a	return -1
73	n/a
74	n/a	def allocate_lock():
75	n/a	"""Dummy implementation of _thread.allocate_lock()."""
76	n/a	return LockType()
77	n/a
78	n/a	def stack_size(size=None):
79	n/a	"""Dummy implementation of _thread.stack_size()."""
80	n/a	if size is not None:
81	n/a	raise error("setting thread stack size not supported")
82	n/a	return 0
83	n/a
84	n/a	def _set_sentinel():
85	n/a	"""Dummy implementation of _thread._set_sentinel()."""
86	n/a	return LockType()
87	n/a
88	n/a	class LockType(object):
89	n/a	"""Class implementing dummy implementation of _thread.LockType.
90	n/a
91	n/a	Compatibility is maintained by maintaining self.locked_status
92	n/a	which is a boolean that stores the state of the lock. Pickling of
93	n/a	the lock, though, should not be done since if the _thread module is
94	n/a	then used with an unpickled ``lock()`` from here problems could
95	n/a	occur from this class not having atomic methods.
96	n/a
97	n/a	"""
98	n/a
99	n/a	def __init__(self):
100	n/a	self.locked_status = False
101	n/a
102	n/a	def acquire(self, waitflag=None, timeout=-1):
103	n/a	"""Dummy implementation of acquire().
104	n/a
105	n/a	For blocking calls, self.locked_status is automatically set to
106	n/a	True and returned appropriately based on value of
107	n/a	``waitflag``. If it is non-blocking, then the value is
108	n/a	actually checked and not set if it is already acquired. This
109	n/a	is all done so that threading.Condition's assert statements
110	n/a	aren't triggered and throw a little fit.
111	n/a
112	n/a	"""
113	n/a	if waitflag is None or waitflag:
114	n/a	self.locked_status = True
115	n/a	return True
116	n/a	else:
117	n/a	if not self.locked_status:
118	n/a	self.locked_status = True
119	n/a	return True
120	n/a	else:
121	n/a	if timeout > 0:
122	n/a	import time
123	n/a	time.sleep(timeout)
124	n/a	return False
125	n/a
126	n/a	__enter__ = acquire
127	n/a
128	n/a	def __exit__(self, typ, val, tb):
129	n/a	self.release()
130	n/a
131	n/a	def release(self):
132	n/a	"""Release the dummy lock."""
133	n/a	# XXX Perhaps shouldn't actually bother to test? Could lead
134	n/a	# to problems for complex, threaded code.
135	n/a	if not self.locked_status:
136	n/a	raise error
137	n/a	self.locked_status = False
138	n/a	return True
139	n/a
140	n/a	def locked(self):
141	n/a	return self.locked_status
142	n/a
143	n/a	def __repr__(self):
144	n/a	return "<%s %s.%s object at %s>" % (
145	n/a	"locked" if self.locked_status else "unlocked",
146	n/a	self.__class__.__module__,
147	n/a	self.__class__.__qualname__,
148	n/a	hex(id(self))
149	n/a	)
150	n/a
151	n/a	# Used to signal that interrupt_main was called in a "thread"
152	n/a	_interrupt = False
153	n/a	# True when not executing in a "thread"
154	n/a	_main = True
155	n/a
156	n/a	def interrupt_main():
157	n/a	"""Set _interrupt flag to True to have start_new_thread raise
158	n/a	KeyboardInterrupt upon exiting."""
159	n/a	if _main:
160	n/a	raise KeyboardInterrupt
161	n/a	else:
162	n/a	global _interrupt
163	n/a	_interrupt = True