/usr/lib/python2.7/dist-packages/asrun/unittest/030_server.py is in code-aster-run 1.13.1-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | #!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import os
import os.path as osp
import re
import unittest
from glob import glob
from common import dict_conf, tmpdir
from data import study_export
from asrun.installation import aster_root
from asrun.run import AsRunFactory
from asrun.profil import AsterProfil
from asrun.core import magic
from asrun.core.server import build_server, build_server_from_profile, TYPES
from asrun.common.sysutils import local_user, local_host
magic.init_logger(filename=osp.join(tmpdir, 'server.log'), debug=True)
class TestServer(unittest.TestCase):
def setUp(self):
self.bstest = "testfile"
self.ftest = osp.join(tmpdir, self.bstest)
open(self.ftest, 'w').write("abcde"*50)
def test01_local(self):
# LocalServer automatically retyped from SSHServer
os.chdir(tmpdir)
jobid = "localserver"
stdid = "loc-%s" % os.getpid()
# test exec
serv = build_server('asrun.plugins.server.SSHServer',
dict_conf['localhost'], dict_conf['localuser'],
aster_root=aster_root,
type=TYPES.EXEC)
assert serv.__class__.__name__ == 'LocalExecServer', serv.__class__
assert serv.host == local_host, serv.host
assert serv.user == local_user, serv.user
assert serv.is_localhost()
res = serv.exec_command(['ls', self.bstest])
assert res[0] == 0, res
assert res[1].find(self.bstest) >= 0, res
assert res[2].strip() == '', res
# test copyto
serv = build_server('asrun.plugins.server.SCPServer',
dict_conf['localhost'], dict_conf['localuser'],
proxy_dir=osp.join(tmpdir, jobid),
type=TYPES.COPY_TO,
jobid=stdid)
assert serv.__class__.__name__ == 'LocalCopyServer', serv.__class__
assert serv.is_localhost()
res = serv.copyto(self.ftest)
assert res == 0, res
fres = glob(osp.join(tmpdir, jobid, "*%s" % stdid, self.bstest))
assert len(fres) == 1 and osp.exists(fres[0]), fres
# test copyfrom
dst = osp.join("from_local", "destdir", self.bstest)
res = serv.copyfrom(dst)
assert res == 0, res
assert osp.isfile(dst)
def test02_ssh(self):
# SSHServer : fails if ssh connection to localhost does not work
os.chdir(tmpdir)
jobid = "sshserver"
stdid = "ssh-%s" % os.getpid()
# test exec
serv = build_server('asrun.plugins.server.SSHServer', '127.0.0.1', dict_conf['localuser'])
assert serv.__class__.__name__ == 'SSHServer', serv.__class__
assert serv.host == "127.0.0.1", serv.host
assert serv.user == dict_conf['localuser'], serv.user
assert serv.get_aster_root() is None
serv.set_aster_root(aster_root)
assert serv.get_aster_root() == aster_root
res = serv.exec_command(['ls', self.ftest])
assert res[0] == 0, res
assert res[1].find(self.bstest) >= 0, res
assert res[2].strip() == '', res
# test copyto
serv = build_server('asrun.plugins.server.SCPServer', '127.0.0.1', dict_conf['localuser'],
proxy_dir=osp.join(tmpdir, jobid),
jobid=stdid)
assert serv.__class__.__name__ == 'SCPServer', serv.__class__
assert serv.get_proxy_dir() is not None
res = serv.copyto(self.ftest)
assert res == 0, res
fres = glob(osp.join(tmpdir, jobid, "*%s" % stdid, self.bstest))
assert len(fres) == 1 and osp.exists(fres[0]), fres
# test copyfrom
dst = osp.join("from_ssh", "destdir", self.bstest)
res = serv.copyfrom(dst)
assert res == 0, res
assert osp.isfile(dst)
def test03_profil(self):
from asrun.common_func import is_localhost2
run = AsRunFactory()
export = osp.join(tmpdir, "object_from_file.export")
open(export, "w").write(study_export % dict_conf)
pr_file = AsterProfil(export)
#XXX should be in data
jobid = "profil_local"
pr_file['protocol_exec'] = 'asrun.plugins.server.SSHServer'
pr_file['protocol_copyto'] = 'asrun.plugins.server.SCPServer'
pr_file['aster_root'] = aster_root
pr_file['proxy_dir'] = osp.join(tmpdir, jobid)
servex = build_server_from_profile(pr_file, TYPES.EXEC)
servcp = build_server_from_profile(pr_file, TYPES.COPY_TO)
assert servex.__class__.__name__ == 'LocalExecServer', serv.__class__
assert servcp.__class__.__name__ == 'LocalCopyServer', serv.__class__
assert servex.get_aster_root() == aster_root
servcp.set_proxy_dir('/tmp')
# force remote server : it can not be forced with is_localhost2
pr_file['serveur'] = '127.0.0.1'
serv = build_server_from_profile(pr_file, TYPES.COPY_TO)
#assert serv.__class__.__name__ == 'SCPServer', serv.__class__
assert is_localhost2(pr_file['serveur'][0])
assert serv.__class__.__name__ == 'LocalCopyServer', serv.__class__
def test04_rsh(self):
# RSHServer : fails if ssh connection to localhost does not work
os.chdir(tmpdir)
jobid = "rshserver"
stdid = "rsh-%s" % os.getpid()
# test exec
serv = build_server('asrun.plugins.server.RSHServer', '127.0.0.1', dict_conf['localuser'])
assert serv.__class__.__name__ == 'RSHServer', serv.__class__
assert serv.host == "127.0.0.1", serv.host
assert serv.user == dict_conf['localuser'], serv.user
assert serv.get_aster_root() is None
serv.set_aster_root(aster_root)
assert serv.get_aster_root() == aster_root
res = serv.exec_command(['ls', self.ftest])
assert res[0] == 0, res
assert res[1].find(self.bstest) >= 0, res
assert res[2].strip() == '', res
# test copyto
serv = build_server('asrun.plugins.server.RCPServer', '127.0.0.1', dict_conf['localuser'],
proxy_dir=osp.join(tmpdir, jobid),
jobid=stdid)
assert serv.__class__.__name__ == 'RCPServer', serv.__class__
assert serv.get_proxy_dir() is not None
res = serv.copyto(self.ftest)
assert res == 0, res
fres = glob(osp.join(tmpdir, jobid, "*%s" % stdid, self.bstest))
assert len(fres) == 1 and osp.exists(fres[0]), fres
# test copyfrom
dst = osp.join("from_rsh", "destdir", self.bstest)
res = serv.copyfrom(dst)
assert res == 0, res
assert osp.isfile(dst)
if __name__ == "__main__":
unittest.main()
|