File: test_xspec.py

package info (click to toggle)
execnet 1.5.0-4
  • links: PTS, VCS
  • area: main
  • in suites: buster
  • size: 744 kB
  • sloc: python: 5,123; makefile: 82; sh: 4
file content (236 lines) | stat: -rw-r--r-- 8,760 bytes parent folder | download
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
import os
import py
import sys
import pytest
import execnet
import subprocess
from execnet.gateway_io import ssh_args, popen_args, vagrant_ssh_args

XSpec = execnet.XSpec


skip_win_pypy = pytest.mark.xfail(condition=hasattr(sys, 'pypy_version_info') and sys.platform.startswith('win'),
                                  reason='failing on Windows on PyPy (#63)')


class TestXSpec:
    def test_norm_attributes(self):
        spec = XSpec("socket=192.168.102.2:8888//python=c:/this/python2.5"
                     "//chdir=d:\hello")
        assert spec.socket == "192.168.102.2:8888"
        assert spec.python == "c:/this/python2.5"
        assert spec.chdir == "d:\hello"
        assert spec.nice is None
        assert not hasattr(spec, '_xyz')

        pytest.raises(AttributeError, "spec._hello")

        spec = XSpec("socket=192.168.102.2:8888//python=python2.5//nice=3")
        assert spec.socket == "192.168.102.2:8888"
        assert spec.python == "python2.5"
        assert spec.chdir is None
        assert spec.nice == "3"

        spec = XSpec("ssh=user@host"
                     "//chdir=/hello/this//python=/usr/bin/python2.5")
        assert spec.ssh == "user@host"
        assert spec.python == "/usr/bin/python2.5"
        assert spec.chdir == "/hello/this"

        spec = XSpec("popen")
        assert spec.popen is True

    def test_ssh_options(self):
        spec = XSpec("ssh=-p 22100 user@host//python=python3")
        assert spec.ssh == "-p 22100 user@host"
        assert spec.python == "python3"

        spec = XSpec(
            "ssh=-i ~/.ssh/id_rsa-passwordless_login -p 22100 user@host"
            "//python=python3")
        assert spec.ssh == \
            "-i ~/.ssh/id_rsa-passwordless_login -p 22100 user@host"
        assert spec.python == "python3"

    def test_execmodel(self):
        spec = XSpec("execmodel=thread")
        assert spec.execmodel == "thread"
        spec = XSpec("execmodel=eventlet")
        assert spec.execmodel == "eventlet"

    def test_ssh_options_and_config(self):
        spec = XSpec("ssh=-p 22100 user@host//python=python3")
        spec.ssh_config = "/home/user/ssh_config"
        assert ssh_args(spec)[:6] == [
            "ssh", "-C", "-F", spec.ssh_config, "-p", "22100"]

    def test_vagrant_options(self):
        spec = XSpec("vagrant_ssh=default//python=python3")
        assert vagrant_ssh_args(spec)[:-1] == [
            'vagrant', 'ssh', 'default', '--', '-C']

    def test_popen_with_sudo_python(self):
        spec = XSpec("popen//python=sudo python3")
        assert popen_args(spec) == [
            'sudo', 'python3', '-u', '-c',
            'import sys;exec(eval(sys.stdin.readline()))'
        ]

    def test_env(self):
        xspec = XSpec("popen//env:NAME=value1")
        assert xspec.env['NAME'] == "value1"

    def test__samefilesystem(self):
        assert XSpec("popen")._samefilesystem()
        assert XSpec("popen//python=123")._samefilesystem()
        assert not XSpec("popen//chdir=hello")._samefilesystem()

    def test__spec_spec(self):
        for x in ("popen", "popen//python=this"):
            assert XSpec(x)._spec == x

    def test_samekeyword_twice_raises(self):
        pytest.raises(ValueError, XSpec, 'popen//popen')
        pytest.raises(ValueError, XSpec, 'popen//popen=123')

    def test_unknown_keys_allowed(self):
        xspec = XSpec("hello=3")
        assert xspec.hello == '3'

    def test_repr_and_string(self):
        for x in ("popen", "popen//python=this"):
            assert repr(XSpec(x)).find("popen") != -1
            assert str(XSpec(x)) == x

    def test_hash_equality(self):
        assert XSpec("popen") == XSpec("popen")
        assert hash(XSpec("popen")) == hash(XSpec("popen"))
        assert XSpec("popen//python=123") != XSpec("popen")
        assert hash(XSpec("socket=hello:8080")) != hash(XSpec("popen"))


class TestMakegateway:
    def test_no_type(self, makegateway):
        pytest.raises(ValueError, lambda: makegateway('hello'))

    @skip_win_pypy
    def test_popen_default(self, makegateway):
        gw = makegateway("")
        assert gw.spec.popen
        assert gw.spec.python is None
        rinfo = gw._rinfo()
        # assert rinfo.executable == sys.executable
        assert rinfo.cwd == os.getcwd()
        assert rinfo.version_info == sys.version_info

    def test_popen_env(self, makegateway):
        gw = makegateway("popen//env:NAME123=123")
        ch = gw.remote_exec("""
            import os
            channel.send(os.environ['NAME123'])
        """)
        value = ch.receive()
        assert value == "123"

    @skip_win_pypy
    def test_popen_explicit(self, makegateway):
        gw = makegateway("popen//python=%s" % sys.executable)
        assert gw.spec.python == sys.executable
        rinfo = gw._rinfo()
        assert rinfo.executable == sys.executable
        assert rinfo.cwd == os.getcwd()
        assert rinfo.version_info == sys.version_info

    def test_popen_cpython25(self, makegateway):
        for trypath in ('python2.5', r'C:\Python25\python.exe'):
            cpython25 = py.path.local.sysfind(trypath)
            if cpython25 is not None:
                cpython25 = cpython25.realpath()
                break
        else:
            pytest.skip("cpython2.5 not found")
        gw = makegateway("popen//python=%s" % cpython25)
        rinfo = gw._rinfo()
        if sys.platform != "darwin":  # it's confusing there
            assert rinfo.executable == cpython25
        assert rinfo.cwd == os.getcwd()
        assert rinfo.version_info[:2] == (2, 5)

    def test_popen_cpython26(self, makegateway):
        for trypath in ('python2.6', r'C:\Python26\python.exe'):
            cpython26 = py.path.local.sysfind(trypath)
            if cpython26 is not None:
                break
        else:
            py.test.skip("cpython2.6 not found")
        gw = makegateway("popen//python=%s" % cpython26)
        rinfo = gw._rinfo()
        # assert rinfo.executable == cpython26
        assert rinfo.cwd == os.getcwd()
        assert rinfo.version_info[:2] == (2, 6)

    @skip_win_pypy
    def test_popen_chdir_absolute(self, testdir, makegateway):
        gw = makegateway("popen//chdir=%s" % testdir.tmpdir)
        rinfo = gw._rinfo()
        assert rinfo.cwd == str(testdir.tmpdir.realpath())

    @skip_win_pypy
    def test_popen_chdir_newsub(self, testdir, makegateway):
        testdir.chdir()
        gw = makegateway("popen//chdir=hello")
        rinfo = gw._rinfo()
        expected = str(testdir.tmpdir.join("hello").realpath()).lower()
        assert rinfo.cwd.lower() == expected

    def test_ssh(self, specssh, makegateway):
        sshhost = specssh.ssh
        gw = makegateway("ssh=%s//id=ssh1" % sshhost)
        rinfo = gw._rinfo()
        assert gw.id == 'ssh1'
        gw2 = execnet.SshGateway(sshhost)
        rinfo2 = gw2._rinfo()
        assert rinfo.executable == rinfo2.executable
        assert rinfo.cwd == rinfo2.cwd
        assert rinfo.version_info == rinfo2.version_info

    @pytest.mark.xfail(reason="bad image name", run=False)
    def test_vagrant(self, makegateway, tmpdir, monkeypatch):
        vagrant = py.path.local.sysfind('vagrant')
        if vagrant is None:
            pytest.skip('Vagrant binary not in PATH')
        monkeypatch.chdir(tmpdir)
        subprocess.check_call("vagrant init hashicorp/precise32", shell=True)
        subprocess.check_call("vagrant up --provider virtualbox", shell=True)
        gw = makegateway("vagrant_ssh=default")
        rinfo = gw._rinfo()
        rinfo.cwd == '/home/vagrant'
        rinfo.executable == '/usr/bin/python'
        subprocess.check_call("vagrant halt", shell=True)

    def test_socket(self, specsocket, makegateway):
        gw = makegateway("socket=%s//id=sock1" % specsocket.socket)
        rinfo = gw._rinfo()
        assert rinfo.executable
        assert rinfo.cwd
        assert rinfo.version_info
        assert gw.id == "sock1"
        # we cannot instantiate a second gateway

    @pytest.mark.xfail(reason='we can\'t instantiate a second gateway')
    def test_socket_second(self, specsocket, makegateway):
        gw = makegateway("socket=%s//id=sock1" % specsocket.socket)
        gw2 = makegateway("socket=%s//id=sock1" % specsocket.socket)
        rinfo = gw._rinfo()
        rinfo2 = gw2._rinfo()
        assert rinfo.executable == rinfo2.executable
        assert rinfo.cwd == rinfo2.cwd
        assert rinfo.version_info == rinfo2.version_info

    def test_socket_installvia(self):
        group = execnet.Group()
        group.makegateway("popen//id=p1")
        gw = group.makegateway("socket//installvia=p1//id=s1")
        assert gw.id == "s1"
        assert gw.remote_status()
        group.terminate()