diff --git a/Lib/test/test_dummy_thread.py b/Lib/test/test_dummy_thread.py --- a/Lib/test/test_dummy_thread.py +++ b/Lib/test/test_dummy_thread.py @@ -11,10 +11,12 @@ import random import unittest from test import support +from unittest import mock DELAY = 0 # Set > 0 when testing a module other than _dummy_thread, such as # the '_thread' module. + class LockTests(unittest.TestCase): """Test lock objects.""" @@ -83,6 +85,19 @@ self.assertGreaterEqual(end_time - start_time, DELAY, "Blocking by unconditional acquiring failed.") + @mock.patch('time.sleep') + def test_acquire_timeout(self, mock_sleep): + """Test invoking acquire() with a positive timeout when the lock is + already acquired. Ensure that time.sleep() is invoked with the given + timeout and that False is returned.""" + + self.lock.acquire() + retval = self.lock.acquire(waitflag=0, timeout=1) + self.assertTrue(mock_sleep.called) + mock_sleep.assert_called_once_with(1) + self.assertEqual(retval, False) + + class MiscTests(unittest.TestCase): """Miscellaneous tests.""" @@ -116,6 +131,7 @@ # KeyboardInterrupt is raised instantly. self.assertRaises(KeyboardInterrupt, _thread.interrupt_main) + class ThreadTests(unittest.TestCase): """Test thread creation.""" @@ -168,15 +184,55 @@ "Not all %s threads executed properly after %s sec." % (thread_count, DELAY)) -def test_main(imported_module=None): - global _thread, DELAY - if imported_module: - _thread = imported_module - DELAY = 2 - if support.verbose: - print() - print("*** Using %s as _thread module ***" % _thread) - support.run_unittest(LockTests, MiscTests, ThreadTests) + def test_args_not_tuple(self): + """ + Test invoking start_new_thread() with a non-tuple value for "args". + Expect TypeError with a meaningful error message to be raised. + """ + with self.assertRaises(TypeError) as cm: + _thread.start_new_thread(mock.Mock(), []) + self.assertEqual(cm.exception.args[0], "2nd arg must be a tuple") -if __name__ == '__main__': - test_main() + def test_kwargs_not_dict(self): + """ + Test invoking start_new_thread() with a non-dict value for "kwargs". + Expect TypeError with a meaningful error message to be raised. + """ + with self.assertRaises(TypeError) as cm: + _thread.start_new_thread(mock.Mock(), tuple(), kwargs=[]) + self.assertEqual(cm.exception.args[0], "3rd arg must be a dict") + + def test_SystemExit(self): + """ + Test invoking start_new_thread() with a function that raises + SystemExit. + The exception should be discarded. + """ + func = mock.Mock(side_effect=SystemExit()) + _thread.start_new_thread(func, tuple()) + + @mock.patch('traceback.print_exc') + def test_RaiseException(self, mock_print_exc): + """ + Test invoking start_new_thread() with a function that raises exception. + + The exception should be discarded and the traceback should be printed + via traceback.print_exc() + """ + func = mock.Mock(side_effect=Exception) + _thread.start_new_thread(func, tuple()) + self.assertTrue(mock_print_exc.called) + + +class TestStackSize(unittest.TestCase): + """Unit tests for _dummy_thread.stack_size""" + + def test_None(self): + retval = _thread.stack_size(None) + self.assertEqual(retval, 0) + + def test_not_None(self): + with self.assertRaises(_thread.error) as cm: + _thread.stack_size("") + self.assertEqual(cm.exception.args[0], + "setting thread stack size not supported")