1import os
2import py
3import sys
4import pytest
5import execnet
6import subprocess
7from execnet.gateway_io import ssh_args, popen_args, vagrant_ssh_args
8
9XSpec = execnet.XSpec
10
11
12skip_win_pypy = pytest.mark.xfail(condition=hasattr(sys, 'pypy_version_info') and sys.platform.startswith('win'),
13                                  reason='failing on Windows on PyPy (#63)')
14
15
16class TestXSpec:
17    def test_norm_attributes(self):
18        spec = XSpec(r"socket=192.168.102.2:8888//python=c:/this/python2.5"
19                     r"//chdir=d:\hello")
20        assert spec.socket == "192.168.102.2:8888"
21        assert spec.python == "c:/this/python2.5"
22        assert spec.chdir == r"d:\hello"
23        assert spec.nice is None
24        assert not hasattr(spec, '_xyz')
25
26        with pytest.raises(AttributeError):
27            spec._hello()
28
29        spec = XSpec("socket=192.168.102.2:8888//python=python2.5//nice=3")
30        assert spec.socket == "192.168.102.2:8888"
31        assert spec.python == "python2.5"
32        assert spec.chdir is None
33        assert spec.nice == "3"
34
35        spec = XSpec("ssh=user@host"
36                     "//chdir=/hello/this//python=/usr/bin/python2.5")
37        assert spec.ssh == "user@host"
38        assert spec.python == "/usr/bin/python2.5"
39        assert spec.chdir == "/hello/this"
40
41        spec = XSpec("popen")
42        assert spec.popen is True
43
44    def test_ssh_options(self):
45        spec = XSpec("ssh=-p 22100 user@host//python=python3")
46        assert spec.ssh == "-p 22100 user@host"
47        assert spec.python == "python3"
48
49        spec = XSpec(
50            "ssh=-i ~/.ssh/id_rsa-passwordless_login -p 22100 user@host"
51            "//python=python3")
52        assert spec.ssh == \
53            "-i ~/.ssh/id_rsa-passwordless_login -p 22100 user@host"
54        assert spec.python == "python3"
55
56    def test_execmodel(self):
57        spec = XSpec("execmodel=thread")
58        assert spec.execmodel == "thread"
59        spec = XSpec("execmodel=eventlet")
60        assert spec.execmodel == "eventlet"
61
62    def test_ssh_options_and_config(self):
63        spec = XSpec("ssh=-p 22100 user@host//python=python3")
64        spec.ssh_config = "/home/user/ssh_config"
65        assert ssh_args(spec)[:6] == [
66            "ssh", "-C", "-F", spec.ssh_config, "-p", "22100"]
67
68    def test_vagrant_options(self):
69        spec = XSpec("vagrant_ssh=default//python=python3")
70        assert vagrant_ssh_args(spec)[:-1] == [
71            'vagrant', 'ssh', 'default', '--', '-C']
72
73    def test_popen_with_sudo_python(self):
74        spec = XSpec("popen//python=sudo python3")
75        assert popen_args(spec) == [
76            'sudo', 'python3', '-u', '-c',
77            'import sys;exec(eval(sys.stdin.readline()))'
78        ]
79
80    def test_env(self):
81        xspec = XSpec("popen//env:NAME=value1")
82        assert xspec.env['NAME'] == "value1"
83
84    def test__samefilesystem(self):
85        assert XSpec("popen")._samefilesystem()
86        assert XSpec("popen//python=123")._samefilesystem()
87        assert not XSpec("popen//chdir=hello")._samefilesystem()
88
89    def test__spec_spec(self):
90        for x in ("popen", "popen//python=this"):
91            assert XSpec(x)._spec == x
92
93    def test_samekeyword_twice_raises(self):
94        pytest.raises(ValueError, XSpec, 'popen//popen')
95        pytest.raises(ValueError, XSpec, 'popen//popen=123')
96
97    def test_unknown_keys_allowed(self):
98        xspec = XSpec("hello=3")
99        assert xspec.hello == '3'
100
101    def test_repr_and_string(self):
102        for x in ("popen", "popen//python=this"):
103            assert repr(XSpec(x)).find("popen") != -1
104            assert str(XSpec(x)) == x
105
106    def test_hash_equality(self):
107        assert XSpec("popen") == XSpec("popen")
108        assert hash(XSpec("popen")) == hash(XSpec("popen"))
109        assert XSpec("popen//python=123") != XSpec("popen")
110        assert hash(XSpec("socket=hello:8080")) != hash(XSpec("popen"))
111
112
113class TestMakegateway:
114    def test_no_type(self, makegateway):
115        pytest.raises(ValueError, lambda: makegateway('hello'))
116
117    @skip_win_pypy
118    def test_popen_default(self, makegateway):
119        gw = makegateway("")
120        assert gw.spec.popen
121        assert gw.spec.python is None
122        rinfo = gw._rinfo()
123        # assert rinfo.executable == sys.executable
124        assert rinfo.cwd == os.getcwd()
125        assert rinfo.version_info == sys.version_info
126
127    @pytest.mark.skipif("not hasattr(os, 'nice')")
128    def test_popen_nice(self, makegateway):
129        gw = makegateway("popen")
130
131        def getnice(channel):
132            import os
133            if hasattr(os, 'nice'):
134                channel.send(os.nice(0))
135            else:
136                channel.send(None)
137        remotenice = gw.remote_exec(getnice).receive()
138        gw.exit()
139        if remotenice is not None:
140            gw = makegateway("popen//nice=5")
141            remotenice2 = gw.remote_exec(getnice).receive()
142            assert remotenice2 == remotenice + 5
143
144    def test_popen_env(self, makegateway):
145        gw = makegateway("popen//env:NAME123=123")
146        ch = gw.remote_exec("""
147            import os
148            channel.send(os.environ['NAME123'])
149        """)
150        value = ch.receive()
151        assert value == "123"
152
153    @skip_win_pypy
154    def test_popen_explicit(self, makegateway):
155        gw = makegateway("popen//python=%s" % sys.executable)
156        assert gw.spec.python == sys.executable
157        rinfo = gw._rinfo()
158        assert rinfo.executable == sys.executable
159        assert rinfo.cwd == os.getcwd()
160        assert rinfo.version_info == sys.version_info
161
162    @skip_win_pypy
163    def test_popen_chdir_absolute(self, testdir, makegateway):
164        gw = makegateway("popen//chdir=%s" % testdir.tmpdir)
165        rinfo = gw._rinfo()
166        assert rinfo.cwd == str(testdir.tmpdir.realpath())
167
168    @skip_win_pypy
169    def test_popen_chdir_newsub(self, testdir, makegateway):
170        testdir.chdir()
171        gw = makegateway("popen//chdir=hello")
172        rinfo = gw._rinfo()
173        expected = str(testdir.tmpdir.join("hello").realpath()).lower()
174        assert rinfo.cwd.lower() == expected
175
176    def test_ssh(self, specssh, makegateway):
177        sshhost = specssh.ssh
178        gw = makegateway("ssh=%s//id=ssh1" % sshhost)
179        rinfo = gw._rinfo()
180        assert gw.id == 'ssh1'
181        gw2 = execnet.SshGateway(sshhost)
182        rinfo2 = gw2._rinfo()
183        assert rinfo.executable == rinfo2.executable
184        assert rinfo.cwd == rinfo2.cwd
185        assert rinfo.version_info == rinfo2.version_info
186
187    @pytest.mark.xfail(reason="bad image name", run=False)
188    def test_vagrant(self, makegateway, tmpdir, monkeypatch):
189        vagrant = py.path.local.sysfind('vagrant')
190        if vagrant is None:
191            pytest.skip('Vagrant binary not in PATH')
192        monkeypatch.chdir(tmpdir)
193        subprocess.check_call("vagrant init hashicorp/precise32", shell=True)
194        subprocess.check_call("vagrant up --provider virtualbox", shell=True)
195        gw = makegateway("vagrant_ssh=default")
196        rinfo = gw._rinfo()
197        rinfo.cwd == '/home/vagrant'
198        rinfo.executable == '/usr/bin/python'
199        subprocess.check_call("vagrant halt", shell=True)
200
201    def test_socket(self, specsocket, makegateway):
202        gw = makegateway("socket=%s//id=sock1" % specsocket.socket)
203        rinfo = gw._rinfo()
204        assert rinfo.executable
205        assert rinfo.cwd
206        assert rinfo.version_info
207        assert gw.id == "sock1"
208        # we cannot instantiate a second gateway
209
210    @pytest.mark.xfail(reason='we can\'t instantiate a second gateway')
211    def test_socket_second(self, specsocket, makegateway):
212        gw = makegateway("socket=%s//id=sock1" % specsocket.socket)
213        gw2 = makegateway("socket=%s//id=sock1" % specsocket.socket)
214        rinfo = gw._rinfo()
215        rinfo2 = gw2._rinfo()
216        assert rinfo.executable == rinfo2.executable
217        assert rinfo.cwd == rinfo2.cwd
218        assert rinfo.version_info == rinfo2.version_info
219
220    def test_socket_installvia(self):
221        group = execnet.Group()
222        group.makegateway("popen//id=p1")
223        gw = group.makegateway("socket//installvia=p1//id=s1")
224        assert gw.id == "s1"
225        assert gw.remote_status()
226        group.terminate()
227