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
127
128
129
130
131
132
133
134
|
"""Raw data collector for Coverage."""
import sys
class PyTracer(object):
"""Python implementation of the raw data tracer."""
# Because of poor implementations of trace-function-manipulating tools,
# the Python trace function must be kept very simple. In particular, there
# must be only one function ever set as the trace function, both through
# sys.settrace, and as the return value from the trace function. Put
# another way, the trace function must always return itself. It cannot
# swap in other functions, or return None to avoid tracing a particular
# frame.
#
# The trace manipulator that introduced this restriction is DecoratorTools,
# which sets a trace function, and then later restores the pre-existing one
# by calling sys.settrace with a function it found in the current frame.
#
# Systems that use DecoratorTools (or similar trace manipulations) must use
# PyTracer to get accurate results. The command-line --timid argument is
# used to force the use of this tracer.
def __init__(self):
# Attributes set from the collector:
self.data = None
self.arcs = False
self.should_trace = None
self.should_trace_cache = None
self.warn = None
# The threading module to use, if any.
self.threading = None
self.cur_file_dict = []
self.last_line = [0]
self.data_stack = []
self.last_exc_back = None
self.last_exc_firstlineno = 0
self.thread = None
self.stopped = False
def __repr__(self):
return "<PyTracer at 0x{0:0x}: {1} lines in {2} files>".format(
id(self),
sum(len(v) for v in self.data.values()),
len(self.data),
)
def _trace(self, frame, event, arg_unused):
"""The trace function passed to sys.settrace."""
if self.stopped:
return
if self.last_exc_back: # TODO: bring this up to speed
if frame == self.last_exc_back:
# Someone forgot a return event.
if self.arcs and self.cur_file_dict:
pair = (self.last_line, -self.last_exc_firstlineno)
self.cur_file_dict[pair] = None
self.cur_file_dict, self.last_line = self.data_stack.pop()
self.last_exc_back = None
if event == 'call':
# Entering a new function context. Decide if we should trace
# in this file.
self.data_stack.append((self.cur_file_dict, self.last_line))
filename = frame.f_code.co_filename
disp = self.should_trace_cache.get(filename)
if disp is None:
disp = self.should_trace(filename, frame)
self.should_trace_cache[filename] = disp
self.cur_file_dict = None
if disp.trace:
tracename = disp.source_filename
if tracename not in self.data:
self.data[tracename] = {}
self.cur_file_dict = self.data[tracename]
# Set the last_line to -1 because the next arc will be entering a
# code block, indicated by (-1, n).
self.last_line = -1
elif event == 'line':
# Record an executed line.
if self.cur_file_dict is not None:
lineno = frame.f_lineno
if self.arcs:
self.cur_file_dict[(self.last_line, lineno)] = None
else:
self.cur_file_dict[lineno] = None
self.last_line = lineno
elif event == 'return':
if self.arcs and self.cur_file_dict:
first = frame.f_code.co_firstlineno
self.cur_file_dict[(self.last_line, -first)] = None
# Leaving this function, pop the filename stack.
self.cur_file_dict, self.last_line = self.data_stack.pop()
elif event == 'exception':
self.last_exc_back = frame.f_back
self.last_exc_firstlineno = frame.f_code.co_firstlineno
return self._trace
def start(self):
"""Start this Tracer.
Return a Python function suitable for use with sys.settrace().
"""
if self.threading:
self.thread = self.threading.currentThread()
sys.settrace(self._trace)
return self._trace
def stop(self):
"""Stop this Tracer."""
self.stopped = True
if self.threading and self.thread != self.threading.currentThread():
# Called on a different thread than started us: we can't unhook
# ourselves, but we've set the flag that we should stop, so we
# won't do any more tracing.
return
if self.warn:
if sys.gettrace() != self._trace:
msg = "Trace function changed, measurement is likely wrong: %r"
self.warn(msg % (sys.gettrace(),))
sys.settrace(None)
def get_stats(self):
"""Return a dictionary of statistics, or None."""
return None
|
