sources for executor.py [rev. unknown]
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
136
""" 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