# Make sure the jobs_config option has been called
if not options.jobs_cfg:
message = _("The option --jobs_config is required\n")
- raise src.SatException( message )
+ logger.write(src.printcolors.printcError(message))
+ return 1
# Find the file in the directories
found = False
if not os.path.exists(local_path):
self.machine.sftp.get(job_path_remote, local_path)
self.remote_log_files.append(local_path)
- except:
- self.err += _("Unable to get %s log file from remote.") % job_path_remote
+ except Exception as e:
+ self.err += _("Unable to get %s log file from remote: %s") % (job_path_remote, str(e))
def has_failed(self):
'''Returns True if the job has failed.
self.err = "TIMEOUT : %s seconds elapsed\n" % str(self.timeout)
try:
self.get_log_files()
- except:
- self.err += _("Unable to get remote log files")
+ except Exception as e:
+ self.err += _("Unable to get remote log files: %s" % e)
def total_duration(self):
return self._Tf - self._T0
# Make sure the jobs_config option has been called
if not options.jobs_cfg:
message = _("The option --jobs_config is required\n")
- raise src.SatException( message )
+ src.printcolors.printcError(message)
+ return 1
# Find the file in the directories
found = False
# Check that the path given by the logs_paths_in_file option
# is a file path that can be written
- if options.logs_paths_in_file:
+ if self.options.logs_paths_in_file:
try:
- dir_file = os.path.dirname(options.logs_paths_in_file)
+ dir_file = os.path.dirname(self.options.logs_paths_in_file)
if not os.path.exists(dir_file):
os.makedirs(dir_file)
- if os.path.exists(options.logs_paths_in_file):
- os.remove(options.logs_paths_in_file)
- file_test = open(options.logs_paths_in_file, "w")
+ if os.path.exists(self.options.logs_paths_in_file):
+ os.remove(self.options.logs_paths_in_file)
+ file_test = open(self.options.logs_paths_in_file, "w")
file_test.close()
except Exception as e:
msg = _("WARNING: the logs_paths_in_file option will "
"not be taken into account.\nHere is the error:")
logger_command.write("%s\n%s\n\n" % (src.printcolors.printcWarning(msg), str(e)))
- options.logs_paths_in_file = None
+ self.options.logs_paths_in_file = None
try:
res = None
res = 1
# If the logs_paths_in_file was called, write the result
# and log files in the given file path
- if options.logs_paths_in_file:
- file_res = open(options.logs_paths_in_file, "w")
+ if self.options.logs_paths_in_file:
+ file_res = open(self.options.logs_paths_in_file, "w")
file_res.write(str(res) + "\n")
for i, filepath in enumerate(logger_command.l_logFiles):
file_res.write(filepath)
seconds = total_time - hours*3600 - minutes*60
# Add the fields corresponding to the end time
# and the total time of command
- endtime = dt.strftime('%d/%Y/%m %Hh%Mm%Ss')
+ endtime = dt.strftime('%Y/%m/%d %Hh%Mm%Ss')
self.xmlFile.append_node_attrib("Site", attrib={"endTime" : endtime})
self.xmlFile.append_node_attrib("Site",
attrib={"TotalTime" : "%ih%im%is" % (hours, minutes, seconds)})
return pid
return 0
-def check_proc_existence_and_kill_multi(regex, nb_kills, time_between_to_checks = 1):
+def check_proc_existence_and_kill_multi(regex, nb_kills, time_between_two_checks = 1):
found = False
i = 0
while not found and i < nb_kills :
found = check_proc_existence_and_kill(regex)
if found:
return found
- time.sleep(time_between_to_checks)
+ time.sleep(time_between_two_checks)
i+=1
return 0
\ No newline at end of file
sys.path.append(os.path.join(testdir, '..', '_testTools'))
from salomeTools import Sat
-from tools import check_proc_existence_and_kill
+from tools import check_proc_existence_and_kill_multi
import HTMLTestRunner
sleep_time = 3
OK = "KO"
sat = Sat("-oUSER.editor='cooledit'")
+ sat.config()
cmd_config = threading.Thread(target=sat.config, args=('-e',))
cmd_config.start()
time.sleep(sleep_time)
editor = sat.cfg.USER.editor
- pid = check_proc_existence_and_kill(editor + ".*" + "salomeTools\.pyconf")
+ pid = check_proc_existence_and_kill_multi(editor + ".*" + "salomeTools\.pyconf", 10)
if pid:
OK = "OK"
OK = "KO"
sat = Sat("-oUSER.editor='cooledit'")
+ sat.config()
cmd_config = threading.Thread(target=sat.config, args=('appli-test -e',))
cmd_config.start()
time.sleep(sleep_time)
editor = sat.cfg.USER.editor
- pid = check_proc_existence_and_kill(editor + ".*" + "appli-test\.pyconf")
+ pid = check_proc_existence_and_kill_multi(editor + ".*" + "appli-test\.pyconf", 10)
if pid:
OK = "OK"
--- /dev/null
+#!/usr/bin/env python
+#-*- coding:utf-8 -*-
+# Copyright (C) 2010-2012 CEA/DEN
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+import unittest
+import os
+import sys
+
+# get execution path
+testdir = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(testdir, '..', '..'))
+sys.path.append(os.path.join(testdir, '..', '_testTools'))
+sys.path.append(os.path.join(testdir, '..', '..','commands'))
+
+from salomeTools import Sat
+import HTMLTestRunner
+
+class TestJob(unittest.TestCase):
+ '''Test of the job command
+ '''
+
+ def test_job(self):
+ '''Test the job command
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ sat.job("--jobs_config .test --job Job 1" )
+
+ ff = open(tmp_file, "r")
+ log_files = ff.readlines()
+ ff.close()
+ os.remove(tmp_file)
+ log_config = [line.replace("\n", "") for line in log_files if 'config.xml' in line]
+
+ text = open(log_config[0], "r").read()
+
+ if "nb_proc" in text:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+
+ def test_failing_job(self):
+ '''Test the job command with a failing command
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.job("--jobs_config .test --job Job 4" )
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_file_conf_not_found(self):
+ '''Test the job command with a wrong file configuration
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.job("--jobs_config NOTEXIST --job Job 4" )
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_no_option_jobs_config(self):
+ '''Test the job command without --jobs_config option
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.job("--job Job 4" )
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_job_not_found(self):
+ '''Test the job command without --jobs_config option
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.job("--jobs_config .test --job NOTEXIST" )
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_description(self):
+ '''Test the sat -h job
+ '''
+
+ OK = "KO"
+
+ import job
+
+ if "Executes the commands of the job defined in the jobs configuration file" in job.description():
+ OK = "OK"
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, "OK")
+
+# test launch
+if __name__ == '__main__':
+ HTMLTestRunner.main()
--- /dev/null
+#!/usr/bin/env python
+#-*- coding:utf-8 -*-
+# Copyright (C) 2010-2012 CEA/DEN
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+import unittest
+import os
+import sys
+
+# get execution path
+testdir = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(testdir, '..', '..'))
+sys.path.append(os.path.join(testdir, '..', '_testTools'))
+sys.path.append(os.path.join(testdir, '..', '..','commands'))
+
+from salomeTools import Sat
+from tools import outRedirection
+import HTMLTestRunner
+
+class TestJobs(unittest.TestCase):
+ '''Test of the jobs command
+ '''
+
+ def test_jobs(self):
+ '''Test the jobs command
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the jobs command
+ sat.jobs("--jobs_config .test --publish" )
+
+ ff = open(tmp_file, "r")
+ log_files = ff.readlines()
+ ff.close()
+ os.remove(tmp_file)
+ log_jobs = [line.replace("\n", "") for line in log_files if 'jobs.xml' in line]
+
+ text = open(log_jobs[0], "r").read()
+
+ expected_res = [
+ "Establishing connection with all the machines",
+ "Executing the jobs",
+ "Results for job"
+ ]
+
+ res = 0
+ for exp_res in expected_res:
+ if exp_res not in text:
+ res += 1
+
+ if res == 0:
+ OK = 'OK'
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_jobs_only_jobs(self):
+ '''Test the jobs command with option --only_jobs
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the jobs command
+ sat.jobs("--jobs_config .test --publish --only_jobs Job 1" )
+
+ ff = open(tmp_file, "r")
+ log_files = ff.readlines()
+ ff.close()
+ os.remove(tmp_file)
+ log_jobs = [line.replace("\n", "") for line in log_files if 'jobs.xml' in line]
+
+ text = open(log_jobs[0], "r").read()
+
+ expected_res = [
+ "Establishing connection with all the machines",
+ "Executing the jobs",
+ "Results for job"
+ ]
+
+ res = 0
+ for exp_res in expected_res:
+ if exp_res not in text:
+ res += 1
+
+ if res == 0:
+ OK = 'OK'
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_no_option_jobs_config(self):
+ '''Test the jobs command without --jobs_config option
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.jobs()
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_file_conf_not_found(self):
+ '''Test the jobs command with a wrong file configuration
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ # Execute the job command
+ res = sat.jobs("--jobs_config NOTEXIST" )
+
+ if res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_option_list(self):
+ '''Test the display of the right value of "sat jobs --list"
+ '''
+
+ OK = "KO"
+
+ # output redirection
+ my_out = outRedirection()
+
+ # The command to test
+ sat = Sat()
+ sat.jobs('--list')
+
+ # stop output redirection
+ my_out.end_redirection()
+
+ # get results
+ res = my_out.read_results()
+
+ # get results
+ if "ERROR" not in res:
+ OK = "OK"
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, "OK")
+
+ def test_description(self):
+ '''Test the sat -h jobs
+ '''
+
+ OK = "KO"
+
+ import jobs
+
+ if "The jobs command launches maintenances that are described in the dedicated jobs configuration file." in jobs.description():
+ OK = "OK"
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, "OK")
+
+# test launch
+if __name__ == '__main__':
+ HTMLTestRunner.main()
sys.path.append(os.path.join(testdir, '..', '..','commands'))
from salomeTools import Sat
-from tools import check_proc_existence_and_kill
+from tools import check_proc_existence_and_kill_multi
import HTMLTestRunner
sleep_time = 2
time.sleep(sleep_time)
browser = sat.cfg.USER.browser
- pid = check_proc_existence_and_kill(browser + ".*" + "xml")
+ pid = check_proc_existence_and_kill_multi(browser + ".*" + "xml", 10)
if pid:
OK = "OK"
time.sleep(sleep_time)
browser = sat.cfg.USER.browser
- check_proc_existence_and_kill(browser + ".*" + "hat\.xml")
+ check_proc_existence_and_kill_multi(browser + ".*" + "hat\.xml", 10)
# Read and check the hat.xml file contains at least one log file corresponding to log
hatFilePath = os.path.join(sat.cfg.SITE.log.log_dir, "hat.xml")
# License along with this library; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+echo "Begin date:"
+date
+echo
+echo "Remove old results ... "
rm -rf .coverage htmlcov
-
+echo "Done"
+echo
+echo "****************************"
coverage run --source=../commands/config.py config/option_value.py > test_res.html
coverage run --source=../commands/config.py -a config/option_value_2.py >> test_res.html
coverage run --source=../commands/config.py -a config/create_user_pyconf.py >> test_res.html
coverage run --source=../commands/config.py,../commands/make.py,../commands/environ.py -a compilation/test_make.py >> test_res.html
coverage run --source=../commands/config.py,../commands/makeinstall.py,../commands/environ.py -a compilation/test_makeinstall.py >> test_res.html
coverage run --source=../commands/config.py,../commands/compile.py,../commands/configure.py,../commands/make.py,../commands/makeinstall.py,../commands/environ.py -a compilation/test_compilation.py >> test_res.html
+coverage run --source=../commands/shell.py -a shell/test_shell.py >> test_res.html
+coverage run --source=../commands/job.py -a job/test_job.py >> test_res.html
+coverage run --source=../commands/jobs.py -a jobs/test_jobs.py >> test_res.html
+echo "****************************"
+echo
+echo "building html coverage"
coverage html
+echo "Done"
+echo
+echo "End date:"
+date
+echo
#firefox test_res.html htmlcov/index.html
--- /dev/null
+#!/usr/bin/env python
+#-*- coding:utf-8 -*-
+# Copyright (C) 2010-2012 CEA/DEN
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public
+# License along with this library; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
+
+import unittest
+import os
+import sys
+
+# get execution path
+testdir = os.path.dirname(os.path.realpath(__file__))
+sys.path.append(os.path.join(testdir, '..', '..'))
+sys.path.append(os.path.join(testdir, '..', '_testTools'))
+sys.path.append(os.path.join(testdir, '..', '..','commands'))
+
+from salomeTools import Sat
+import HTMLTestRunner
+
+class TestShell(unittest.TestCase):
+ '''Test of the shell command
+ '''
+
+ def test_shell(self):
+ '''Test the shell command with the --command option
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ sat.config()
+ sat_way = sat.cfg.VARS.salometoolsway
+
+ # Execute the shell command
+ sat.shell("--command ls " + sat_way)
+
+ ff = open(tmp_file, "r")
+ log_files = ff.readlines()
+ ff.close()
+ os.remove(tmp_file)
+ log_files = [line.replace("\n", "") for line in log_files]
+
+ text = open(log_files[2], "r").read()
+
+ if "salomeTools.py" in text:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_failing_shell(self):
+ '''Test the shell command with the --command option with a failing command
+ '''
+ OK = 'KO'
+ tmp_file = "/tmp/test.txt"
+
+ sat = Sat("-l " + tmp_file)
+
+ sat.config()
+
+ # Execute the shell command
+ res = sat.shell("--command i_fail")
+
+ ff = open(tmp_file, "r")
+ log_files = ff.readlines()
+ ff.close()
+ os.remove(tmp_file)
+ log_files = [line.replace("\n", "") for line in log_files]
+
+ text = open(log_files[2], "r").read()
+
+ if "i_fail" in text and res == 1:
+ OK = 'OK'
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, 'OK')
+
+ def test_description(self):
+ '''Test the sat -h shell
+ '''
+
+ OK = "KO"
+
+ import shell
+
+ if "Executes the shell command passed as argument" in shell.description():
+ OK = "OK"
+
+ # pyunit method to compare 2 str
+ self.assertEqual(OK, "OK")
+
+# test launch
+if __name__ == '__main__':
+ HTMLTestRunner.main()
