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
135
|
""" Remote executor
"""
import py, os, sys
from py.__.test.rsession.outcome import Outcome, ReprOutcome
from py.__.test.rsession.box import Box
from py.__.test.rsession import repevent
from py.__.test.outcome import Skipped, Failed
class RunExecutor(object):
""" Same as in executor, but just running run
"""
wraps = False
def __init__(self, item, usepdb=False, reporter=None, config=None):
self.item = item
self.usepdb = usepdb
self.reporter = reporter
self.config = config
assert self.config
def run(self, capture=True):
if capture:
self.item.startcapture()
try:
self.item.run()
finally:
self.item.finishcapture()
else:
self.item.run()
def execute(self, capture=True):
try:
self.run(capture)
outcome = Outcome()
except Skipped, e:
outcome = Outcome(skipped=str(e))
except (SystemExit, KeyboardInterrupt):
raise
except:
e = sys.exc_info()[1]
if isinstance(e, Failed) and e.excinfo:
excinfo = e.excinfo
else:
excinfo = py.code.ExceptionInfo()
if isinstance(self.item, py.test.collect.Function):
fun = self.item.obj # hope this is stable
code = py.code.Code(fun)
excinfo.traceback = excinfo.traceback.cut(
path=code.path, firstlineno=code.firstlineno)
outcome = Outcome(excinfo=excinfo, setupfailure=False)
if self.usepdb:
if self.reporter is not None:
self.reporter(repevent.ImmediateFailure(self.item,
ReprOutcome(outcome.make_repr
(self.config.option.tbstyle))))
import pdb
pdb.post_mortem(excinfo._excinfo[2])
# XXX hmm, we probably will not like to continue from that
# point
raise SystemExit()
outcome.stdout, outcome.stderr = self.item._getouterr()
return outcome
class ApigenExecutor(RunExecutor):
""" Same as RunExecutor, but takes tracer to trace calls as
an argument to execute
"""
def execute(self, tracer):
self.tracer = tracer
return super(ApigenExecutor, self).execute()
def wrap_underlaying(self, target, *args):
try:
self.tracer.start_tracing()
return target(*args)
finally:
self.tracer.end_tracing()
def run(self, capture):
""" We want to trace *only* function objects here. Unsure
what to do with custom collectors at all
"""
if hasattr(self.item, 'obj') and type(self.item) is py.test.collect.Function:
self.item.execute = self.wrap_underlaying
self.item.run()
class BoxExecutor(RunExecutor):
""" Same as RunExecutor, but boxes test instead
"""
wraps = True
def execute(self):
def fun():
outcome = RunExecutor.execute(self, False)
return outcome.make_repr(self.config.option.tbstyle)
b = Box(fun, config=self.config)
pid = b.run()
assert pid
if b.retval is not None:
passed, setupfailure, excinfo, skipped, critical, _, _, _\
= b.retval
return (passed, setupfailure, excinfo, skipped, critical, 0,
b.stdoutrepr, b.stderrrepr)
else:
return (False, False, None, False, False, b.signal,
b.stdoutrepr, b.stderrrepr)
class AsyncExecutor(RunExecutor):
""" same as box executor, but instead it returns function to continue
computations (more async mode)
"""
wraps = True
def execute(self):
def fun():
outcome = RunExecutor.execute(self, False)
return outcome.make_repr(self.config.option.tbstyle)
b = Box(fun, config=self.config)
parent, pid = b.run(continuation=True)
def cont(waiter=os.waitpid):
parent(pid, waiter=waiter)
if b.retval is not None:
passed, setupfailure, excinfo, skipped,\
critical, _, _, _ = b.retval
return (passed, setupfailure, excinfo, skipped, critical, 0,
b.stdoutrepr, b.stderrrepr)
else:
return (False, False, None, False, False,
b.signal, b.stdoutrepr, b.stderrrepr)
return cont, pid
|
