ยปCore Development>Code coverage>Lib/test/test_dummy_thread.py

Python code coverage for Lib/test/test_dummy_thread.py

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