Commit 670f521a authored by Ralf Schmitt's avatar Ralf Schmitt

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.
parent 5994064a
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)
......@@ -317,35 +317,6 @@ class ThreadTests(unittest.TestCase):
""" % 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
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment