Commits

Ralf Schmitt committed d951cc6

move test_finalize_with_trace to test__threading_vs_settrace.py

explicitly test that sys.settrace isn't called and test the same
issue when a non-main thread exits.

Comments (0)

Files changed (2)

greentest/test__threading_vs_settrace.py

+from __future__ import with_statement
+
+import sys
+import subprocess
+import unittest
+import gevent.thread
+
+script = """
+from gevent import monkey
+monkey.patch_all()
+import sys, os, threading, time
+
+
+# A deadlock-killer, to prevent the
+# testsuite to hang forever
+def killer():
+    time.sleep(0.1)
+    sys.stdout.write('..program blocked; aborting!')
+    sys.stdout.flush()
+    os._exit(2)
+t = threading.Thread(target=killer)
+t.daemon = True
+t.start()
+
+
+def trace(frame, event, arg):
+    if threading is not None:
+        threading.current_thread()
+    return trace
+
+
+def doit():
+    sys.stdout.write("..thread started..")
+
+
+def test1():
+    t = threading.Thread(target=doit)
+    t.start()
+    t.join()
+    sys.settrace(None)
+
+sys.settrace(trace)
+if len(sys.argv) > 1:
+    test1()
+
+sys.stdout.write("..finishing..")
+"""
+
+
+class ThreadTrace(unittest.TestCase):
+    def test_untraceable_lock(self):
+        old = sys.gettrace()
+        lst = []
+        try:
+            def trace(frame, ev, arg):
+                lst.append((frame.f_code.co_filename, frame.f_lineno, ev))
+                print "TRACE: %s:%s %s" % lst[-1]
+                return trace
+
+            with gevent.thread.allocate_lock():
+                sys.settrace(trace)
+        finally:
+            sys.settrace(old)
+
+        self.failUnless(lst == [], "trace not empty")
+
+    def run_script(self, more_args=[]):
+        rc = subprocess.call([sys.executable, "-c", script] + more_args)
+        self.failIf(rc == 2, "interpreter was blocked")
+        self.failUnless(rc == 0, "Unexpected error")
+
+    def test_finalize_with_trace(self):
+        self.run_script()
+
+    def test_bootstrap_inner_with_trace(self):
+        self.run_script(["1"])
+
+
+if __name__ == "__main__":
+    import test.test_support
+    test.test_support.run_unittest(ThreadTrace)

greentest/test_threading_2.py

                 """ % setup_4])
             self.assertEqual(rc, 42)
 
-    def test_finalize_with_trace(self):
-        # Issue1733757
-        # Avoid a deadlock when sys.settrace steps into threading._shutdown
-        import subprocess
-        rc = subprocess.call([sys.executable, "-c", """if 1:
-%s
-            import sys, threading
-
-            # A deadlock-killer, to prevent the
-            # testsuite to hang forever
-            def killer():
-                import os, time
-                time.sleep(2)
-                print 'program blocked; aborting'
-                os._exit(2)
-            t = threading.Thread(target=killer)
-            t.daemon = True
-            t.start()
-
-            # This is the trace function
-            def func(frame, event, arg):
-                threading.current_thread()
-                return func
-
-            sys.settrace(func)
-            """ % setup_3])
-        self.failIf(rc == 2, "interpreted was blocked")
-        self.failUnless(rc == 0, "Unexpected error")
-
     if sys.version_info[:2] > (2, 5):
         def test_join_nondaemon_on_shutdown(self):
             # Issue 1722344