sources for test_remote.py [rev. 38799]
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
import py
try:
    from py.magic import greenlet
except (ImportError, RuntimeError), e:
    py.test.skip(str(e))
class RGreenletBunch:
    def __init__(self, gateway):
        self.channel = gateway.remote_exec('''
            from py.magic import greenlet
            glob = {"greenlet": greenlet}
            gids = {}
            while True:
                key, code, args = channel.receive()
                if args is not None:
                    if code is not None:
                        def run(code=code):
                            exec code in glob, {}
                        gids[key] = greenlet(run)
                    result = gids[key].switch(*args)
                    channel.send(result)
                else:
                    del gids[key]
        ''')
    def greenlet(self, code):
        return RGreenlet(self, code)
class RGreenlet:
    def __init__(self, bunch, code):
        self.channel = bunch.channel
        self.code    = str(py.code.Source(code))
    def switch(self, *args):
        self.channel.send((id(self), self.code, args))
        self.code = None     # only send over the code the first time
        return self.channel.receive()
    def __del__(self):
        if self.code is None:
            self.channel.send((id(self), None, None))
def test_rgreenlet():
    gw = py.execnet.PopenGateway()
    bunch = RGreenletBunch(gw)
    g = bunch.greenlet('''
        x = greenlet.getcurrent().parent.switch(42)
        y = greenlet.getcurrent().parent.switch(x+1)
        greenlet.getcurrent().parent.switch(y+2)
        import os
        greenlet.getcurrent().parent.switch(os.getpid())
    ''')
    result = g.switch()
    assert result == 42
    result = g.switch(102)
    assert result == 103
    result = g.switch(-93)
    assert result == -91
    import os
    result = g.switch()
    assert result != os.getpid()