1# If stdlib threading is imported *BEFORE* monkey patching, *and* 2# there is a native thread created, we can still get the current 3# (main) thread, and it's not a DummyThread. 4# Joining the native thread also does not fail 5 6import threading 7from time import sleep as time_sleep 8 9import gevent.testing as greentest 10 11class NativeThread(threading.Thread): 12 do_run = True 13 14 def run(self): 15 while self.do_run: 16 time_sleep(0.1) 17 18 def stop(self, timeout=None): 19 self.do_run = False 20 self.join(timeout=timeout) 21 22native_thread = None 23 24class Test(greentest.TestCase): 25 26 @classmethod 27 def tearDownClass(cls): 28 global native_thread 29 if native_thread is not None: 30 native_thread.stop(1) 31 native_thread = None 32 33 def test_main_thread(self): 34 current = threading.current_thread() 35 self.assertNotIsInstance(current, threading._DummyThread) 36 self.assertIsInstance(current, monkey.get_original('threading', 'Thread')) 37 # in 3.4, if the patch is incorrectly done, getting the repr 38 # of the thread fails 39 repr(current) 40 41 if hasattr(threading, 'main_thread'): # py 3.4 42 self.assertEqual(threading.current_thread(), threading.main_thread()) 43 44 @greentest.ignores_leakcheck # because it can't be run multiple times 45 def test_join_native_thread(self): 46 if native_thread is None or not native_thread.do_run: # pragma: no cover 47 self.skipTest("native_thread already closed") 48 49 self.assertTrue(native_thread.is_alive()) 50 51 native_thread.stop(timeout=1) 52 self.assertFalse(native_thread.is_alive()) 53 54 # again, idempotent 55 native_thread.stop() 56 self.assertFalse(native_thread.is_alive()) 57 58 59if __name__ == '__main__': 60 native_thread = NativeThread() 61 native_thread.daemon = True 62 native_thread.start() 63 64 # Only patch after we're running 65 from gevent import monkey 66 monkey.patch_all() # pragma: testrunner-no-monkey-combine 67 68 greentest.main() 69