/usr/lib/codeaster/asrun/unittest/020_object.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 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 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 | #!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys
import os
import os.path as osp
import re
import unittest
from common import dict_conf, tmpdir, aster_version
from data import study_export, available_hosts
from asrun.installation import aster_root
from asrun.run import AsRunFactory
from asrun.profil import AsterProfil, MODES, ExportEntry
from asrun.config import AsterConfig, build_config_from_export, build_config_of_version
from asrun.maintenance import get_aster_version
from asrun.repart import (ResourceManager, PureLocalResource,
ALLOCATED, NORESOURCE, OVERLIMIT)
from asrun.common.rcfile import parse_config
from asrun.common.sysutils import local_user, local_full_host, is_localhost, \
is_localuser, get_home_directory
from asrun.common.utils import YES_VALUES
remhost = dict_conf['remhost']
class TestObjects(unittest.TestCase):
def test01_run(self):
run = AsRunFactory()
assert type(run.get_as_run_cmd()) is list and len(run.get_as_run_cmd()) > 0
p1 = run.get_pid()
p2 = run.get_pid()
assert p1 != p2
args = run.get_remote_args()
assert args[0].find("remote_shell") > -1 and args[1].find("remote_copy") > -1
# test GetGrav
assert run.GetGrav('NO_TEST_RESU') == -1
assert run.GetGrav('OK') == 0
assert run.GetGrav('<A>_ALARM') == 1
assert run.GetGrav('NOOK_TEST_RESU') == 2
assert run.GetGrav('<S>_CPU_LIMIT') == 4
assert run.GetGrav('NO_RESU_FILE') == 6
assert run.GetGrav('<F>_ERROR') == 10
assert run.GetHostName() == local_full_host
# test get_version_path
rgvp = run.get_version_path(aster_version)
absp = osp.join(aster_root, aster_version)
if rgvp != absp:
print "\nWARNING : %s != %s : " \
"should fail only if version is not under %s." % (rgvp, absp, aster_root)
rgvp = run.get_version_path(aster_version, "/aster")
absp = osp.join("/aster", aster_version)
if rgvp != absp:
print "\nWARNING : %s != %s : " \
"should fail only if version is not under %s." % (rgvp, absp, aster_root)
def test02a_profil(self):
run = AsRunFactory()
export = osp.join(tmpdir, "object_from_file.export")
cnt = study_export % dict_conf + os.linesep + \
"F libr $HOME/filename.txt D 0"
open(export, "w").write(cnt)
pr_cnt = AsterProfil()
pr_cnt.parse(cnt)
pr_cnt.set_filename(osp.join(tmpdir, "object_from_content.export"))
pr_file = AsterProfil(export)
# the same + run object
pr_run = AsterProfil(export, run)
for prof in (pr_file, pr_cnt, pr_run):
assert type(prof['nomjob']) in (list, tuple), prof.get_filename()
assert prof['nomjob'][0] == "study", prof.get_filename()
# undefined field
assert type(prof['display']) in (list, tuple), prof.get_filename()
assert prof['display'][0].strip() == "", prof.get_filename()
vref = aster_version
if prof.run is not None:
vref = run.get_version_path(vref)
assert prof.get_version_path() == vref, prof.get_filename()
llibr = prof.get_type('libr')
assert len(llibr) == 1, len(llibr)
assert llibr[0].path == osp.join(get_home_directory(), 'filename.txt'), llibr[0].path
assert llibr[0].path.find('$') == -1, llibr[0].path
pr_cnt.WriteExportTo(pr_cnt.get_filename())
reloc = pr_cnt.copy()
reloc.relocate(local_full_host, "/tmp")
reloc.WriteExportTo(osp.join(tmpdir, "relocated.export"))
def test02b_profil_remote(self):
run = AsRunFactory()
export = osp.join(tmpdir, "object_from_file.export")
open(export, "w").write(study_export % dict_conf)
pr_run = AsterProfil(export, run)
comm0 = pr_run.data[0]['path']
pr_run.from_remote_server()
# check difference
comm1 = pr_run.data[0]['path']
diff = comm1.replace(comm0, "")
assert diff == "%s@%s:" % (local_user, local_full_host), (comm0, comm1, ">>>", diff)
all = pr_run.get_data()
assert len(all) == 1
loc = all.get_on_serv(local_full_host)
assert len(loc) == 1
entry = loc[0]
assert is_localhost(entry.host) and is_localuser(entry.user), entry.repr()
if len(available_hosts) > 0:
oth = all.get_on_serv(remhost)
assert len(oth) == 0, (oth, remhost)
res = pr_run.get_result()
assert len(res) == 0, res
def test02c_mode(self):
run = AsRunFactory()
export = osp.join(tmpdir, "object_from_file.export")
open(export, "w").write(study_export % dict_conf)
pr_run = AsterProfil(export, run)
pr_run.set_running_mode(MODES.BATCH)
vref = run.get("batch") in YES_VALUES and "batch" or "interactif"
assert pr_run["mode"][0] == vref, pr_run["mode"]
# string assignement is used by the OM module
pr_run.set_running_mode("interactif")
assert pr_run["mode"][0] == "interactif", pr_run["mode"]
def test02d_profil_win(self):
run = AsRunFactory()
prwin = AsterProfil(osp.join("datafiles", "with_win_CR.export"))
assert prwin.args.get("memjeveux", "").isdigit()
assert prwin['actions'][0] == "make_etude"
assert len(prwin.Get("D", "comm")) > 0
#path = prwin.Get("D", "comm")[0]["path"]
#spl = split_path(path)
#assert len(spl) == 9 and spl[0] == "C:"
def test02e_relocate(self):
from test_relocate import main
assert len(available_hosts) > 0, "no remote host available."
sample = osp.join("datafiles", "sample.export")
txt = open(sample, 'r').read() % dict_conf
export = osp.join(tmpdir, "test02e.export")
open(export, "w").write(txt)
iret, prof = main(export)
assert iret == 0, iret
# check that basename are all different after relocation
lf, lo = prof.get_type('f', with_completion=True)
assert len(lf) == 2, lf
ltyp = prof.get_type('libr')
assert len(ltyp) == 4, ltyp
llibr = lo.get_type('libr')
assert len(llibr) == 4, llibr
bname = set([osp.basename(e.path) for e in ltyp])
assert len(bname) == 4, bname
# check that the same entry is not duplicated
lc = prof.get_type('c')
assert len(lc) == 1, lc
# executable must not appear in datas
ldat = prof.get_data().get_type('exec')
assert len(ldat) == 0, ldat
lres = prof.get_result().get_type('exec')
assert len(lres) == 1, lres
assert lres[0].data, lres[0]
def test02f_relocate(self):
run = AsRunFactory()
sample = osp.join("datafiles", "sample2.export")
txt = open(sample, 'r').read() % dict_conf
prof = AsterProfil()
prof.parse(txt)
# relocate results
results = prof.get_result()
newdir = "$HOME/%s" % 'multi.NUMBER'
prof.relocate(serv=remhost, newdir=newdir, fromlist=results)
lf = prof.get_type('resu_test')
assert len(lf) == 1, len(lf)
assert lf[0].get_path_text().find('$HOME') > -1, lf[0].path
def test02g_entry(self):
# test support of environment variables
fname = ExportEntry("$HOME/fname_$LOGNAME.ext")
assert fname.path.count('$') == 0, fname.path
if len(available_hosts) > 0:
rhost = available_hosts.keys()[0]
fname = ExportEntry("%s:$HOME/fname_$LOGNAME.ext" % rhost)
assert fname.path.count('$') == 2, fname.path
# check set method
fname.set_pathname("%s:$HOME/fname_$LOGNAME.ext" % rhost)
assert fname.path.find(':') < 0, fname.path
def test02h_entry_collection(self):
#check that we can change values into EntryExport objects
# even if _compatibility must be called
run = AsRunFactory()
sample = osp.join("datafiles", "sample2.export")
prof = AsterProfil(sample, run)
for entr in prof.get_data():
entr.path = osp.basename(entr.path) + '.changed'
prof._compatibility()
assert prof.get_data()[0].path.endswith('.changed'), prof
def test03_config(self):
run = AsRunFactory()
vers = dict_conf['ASTER_VERSION']
assert vers is not None, "default_vers is None! Check aster configuration file"
verspath = run.get_version_path(vers)
# 1. by hand
ficconf = osp.join(verspath, 'config.txt')
conf1 = AsterConfig(ficconf, run)
# 2. for a version
conf2 = build_config_of_version(run, vers)
# 3. from export
prof = AsterProfil()
prof.add( ExportEntry(ficconf, type='conf', data=True) )
conf3 = build_config_from_export(run, prof)
# test
valref = osp.join(verspath, 'bibfor')
assert conf1.get_with_absolute_path('SRCFOR')[0] == valref, \
conf1.get_with_absolute_path('SRCFOR')[0]
assert conf2.get_with_absolute_path('SRCFOR')[0] == valref, \
conf2.get_with_absolute_path('SRCFOR')[0]
assert conf3.get_with_absolute_path('SRCFOR')[0] == valref, \
conf3.get_with_absolute_path('SRCFOR')[0]
def test04_build(self):
pass
def test05_resource(self):
run = AsRunFactory()
rc = PureLocalResource(run)
result = rc.CheckHosts(run, silent=True)
assert result == (1, 1), "local host should be alone and ok : %s" % repr(result)
hosts = rc.get_all_connected_hosts()
info = rc.GetConfig(hosts[0])
cputot = info["cpu"]
assert rc.get_sum("cpu") == cputot, rc.get_sum("cpu")
cpu_i = max(cputot / 2, 1)
host, status = rc.Request(nomjob="job_1", cpu=cpu_i, run=run)
assert status is ALLOCATED, status
if cputot >= 2:
host, status = rc.Request(nomjob="job_2", cpu=cpu_i, run=run)
assert status is ALLOCATED, status
host, status = rc.Request(nomjob="job_3", cpu=cpu_i, run=run)
assert status is NORESOURCE, status
host, status = rc.Request(nomjob="job_4", cpu=cputot*2, run=run)
assert status is OVERLIMIT, status
rc.Free("job_1")
rc.Free("job_2")
txt = rc.Load()
assert type(txt) in (str, unicode)
txt = rc.repr_history()
assert type(txt) in (str, unicode)
# create a new ResourceManager object from the previous one
content = rc.hostinfo_repr()
hostfile = osp.join(tmpdir, "hostfile_from_purelocal")
open(hostfile, "w").write(content)
host_infos = parse_config(hostfile)
newrc = ResourceManager(host_infos)
hosts = newrc.get_all_connected_hosts()
assert len(hosts) == 1
def test06_maint(self):
run = AsRunFactory()
vers = dict_conf['ASTER_VERSION']
assert vers is not None, "default_vers is None! Check aster configuration file"
res = get_aster_version(vers)
assert type(res) in (list, tuple) and len(res) == 5, res
if __name__ == "__main__":
unittest.main()
|