2 from unittest import TestCase
4 from eventlet import debug
5 from eventlet.support import capture_stderr, six
6 from tests import LimitedTestCase, main
10 class TestSpew(TestCase):
12 self.orig_trace = sys.settrace
13 sys.settrace = self._settrace
17 sys.settrace = self.orig_trace
18 sys.stdout = sys.__stdout__
20 def _settrace(self, cb):
25 assert isinstance(self.tracer, debug.Spew)
27 def test_unspew(self):
30 assert self.tracer is None
33 sys.stdout = six.StringIO()
37 lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
38 output = sys.stdout.getvalue()
39 assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
40 assert "f=<frame object at" in output
42 def test_line_nofile(self):
43 sys.stdout = six.StringIO()
47 f = eval("sys._getframe()", g)
50 output = sys.stdout.getvalue()
51 assert "[unknown]:%i" % lineno in output, "Didn't find [unknown]:%i in %s" % (lineno, output)
52 assert "VM instruction #" in output, output
54 def test_line_global(self):
56 sys.stdout = six.StringIO()
57 GLOBAL_VAR = debug.Spew()
59 GLOBAL_VAR(f, "line", None)
60 lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
61 output = sys.stdout.getvalue()
62 assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
63 assert "f=<frame object at" in output
64 assert "GLOBAL_VAR" in f.f_globals
65 assert "GLOBAL_VAR=<eventlet.debug.Spew object at" in output
68 def test_line_novalue(self):
69 sys.stdout = six.StringIO()
70 s = debug.Spew(show_values=False)
73 lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
74 output = sys.stdout.getvalue()
75 assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
76 assert "f=<frame object at" not in output
78 def test_line_nooutput(self):
79 sys.stdout = six.StringIO()
80 s = debug.Spew(trace_names=['foo'])
83 output = sys.stdout.getvalue()
87 class TestDebug(LimitedTestCase):
88 def test_everything(self):
89 debug.hub_exceptions(True)
90 debug.hub_exceptions(False)
91 debug.tpool_exceptions(True)
92 debug.tpool_exceptions(False)
93 debug.hub_listener_stacks(True)
94 debug.hub_listener_stacks(False)
95 debug.hub_timer_stacks(True)
96 debug.hub_timer_stacks(False)
97 debug.format_hub_listeners()
98 debug.format_hub_timers()
100 def test_hub_exceptions(self):
101 debug.hub_exceptions(True)
102 server = eventlet.listen(('0.0.0.0', 0))
103 client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
104 client_2, addr = server.accept()
110 with capture_stderr() as fake:
111 gt = eventlet.spawn(hurl, client_2)
115 # allow the "hurl" greenlet to trigger the KeyError
116 # not sure why the extra context switch is needed
118 self.assertRaises(KeyError, gt.wait)
119 debug.hub_exceptions(False)
120 # look for the KeyError exception in the traceback
121 assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()
123 if __name__ == "__main__":