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