1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
import io
import os
import sys
from eventlet import debug
import tests
import eventlet
class TestSpew(tests.LimitedTestCase):
def setUp(self):
self.orig_trace = sys.settrace
sys.settrace = self._settrace
self.tracer = None
def tearDown(self):
sys.settrace = self.orig_trace
sys.stdout = sys.__stdout__
def _settrace(self, cb):
self.tracer = cb
def test_spew(self):
debug.spew()
assert isinstance(self.tracer, debug.Spew)
def test_unspew(self):
debug.spew()
debug.unspew()
assert self.tracer is None
def test_line(self):
frame_str = "f=<frame at"
sys.stdout = io.StringIO()
s = debug.Spew()
f = sys._getframe()
s(f, "line", None)
lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
output = sys.stdout.getvalue()
assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
assert frame_str in output
def test_line_nofile(self):
sys.stdout = io.StringIO()
s = debug.Spew()
g = globals().copy()
del g['__file__']
f = eval("sys._getframe()", g)
lineno = f.f_lineno
s(f, "line", None)
output = sys.stdout.getvalue()
assert "[unknown]:%i" % lineno in output, "Didn't find [unknown]:%i in %s" % (lineno, output)
if "PYTEST_XDIST_WORKER" not in os.environ:
assert "VM instruction #" in output, output
def test_line_global(self):
frame_str = "f=<frame at"
global GLOBAL_VAR
sys.stdout = io.StringIO()
GLOBAL_VAR = debug.Spew()
f = sys._getframe()
GLOBAL_VAR(f, "line", None)
lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
output = sys.stdout.getvalue()
assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
assert frame_str in output
assert "GLOBAL_VAR" in f.f_globals
assert "GLOBAL_VAR=<eventlet.debug.Spew object at" in output
del GLOBAL_VAR
def test_line_novalue(self):
sys.stdout = io.StringIO()
s = debug.Spew(show_values=False)
f = sys._getframe()
s(f, "line", None)
lineno = f.f_lineno - 1 # -1 here since we called with frame f in the line above
output = sys.stdout.getvalue()
assert "%s:%i" % (__name__, lineno) in output, "Didn't find line %i in %s" % (lineno, output)
assert "f=<frame object at" not in output
def test_line_nooutput(self):
sys.stdout = io.StringIO()
s = debug.Spew(trace_names=['foo'])
f = sys._getframe()
s(f, "line", None)
output = sys.stdout.getvalue()
assert output == ""
class TestDebug(tests.LimitedTestCase):
def test_everything(self):
debug.hub_exceptions(True)
debug.hub_exceptions(False)
debug.tpool_exceptions(True)
debug.tpool_exceptions(False)
debug.hub_listener_stacks(True)
debug.hub_listener_stacks(False)
debug.hub_timer_stacks(True)
debug.hub_timer_stacks(False)
debug.format_hub_listeners()
debug.format_hub_timers()
def test_hub_exceptions(self):
debug.hub_exceptions(True)
server = eventlet.listen(('0.0.0.0', 0))
client = eventlet.connect(('127.0.0.1', server.getsockname()[1]))
client_2, addr = server.accept()
def hurl(s):
s.recv(1)
{}[1] # keyerror
with tests.capture_stderr() as fake:
gt = eventlet.spawn(hurl, client_2)
eventlet.sleep(0.001)
client.send(b' ')
eventlet.sleep(0.001)
# allow the "hurl" greenlet to trigger the KeyError
# not sure why the extra context switch is needed
eventlet.sleep(0.001)
self.assertRaises(KeyError, gt.wait)
debug.hub_exceptions(False)
# look for the KeyError exception in the traceback
assert 'KeyError: 1' in fake.getvalue(), "Traceback not in:\n" + fake.getvalue()
|
