src/test_module.py

   1 #!/usr/bin/env python
   2 #-*- coding:utf-8 -*-
   3 #  Copyright (C) 2010-2013  CEA/DEN
   4 #
   5 #  This library is free software; you can redistribute it and/or
   6 #  modify it under the terms of the GNU Lesser General Public
   7 #  License as published by the Free Software Foundation; either
   8 #  version 2.1 of the License.
   9 #
  10 #  This library is distributed in the hope that it will be useful,
  11 #  but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13 #  Lesser General Public License for more details.
  14 #
  15 #  You should have received a copy of the GNU Lesser General Public
  16 #  License along with this library; if not, write to the Free Software
  17 #  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307 USA
  18
  19 # Python 2/3 compatibility for execfile function
  20 try:
  21     execfile
  22 except:
  23     def execfile(somefile, global_vars, local_vars):
  24         with open(somefile) as f:
  25             code = compile(f.read(), somefile, 'exec')
  26             exec(code, global_vars, local_vars)
  27
  28
  29 import os
  30 import sys
  31 import datetime
  32 import shutil
  33 import string
  34 import imp
  35 import subprocess
  36
  37 from . import fork
  38 import src
  39
  40 # directories not considered as test grids
  41 C_IGNORE_GRIDS = ['.git', '.svn', 'RESSOURCES']
  42
  43 # Get directory to be used for the temporary files.
  44 #
  45 def getTmpDirDEFAULT():
  46     if src.architecture.is_windows():
  47         directory = os.getenv("TEMP")
  48     else:
  49         # for Linux: use /tmp/logs/{user} folder
  50         directory = os.path.join( '/tmp', 'logs', os.getenv("USER", "unknown"))
  51     return directory
  52
  53 class Test:
  54     def __init__(self,
  55                  config,
  56                  logger,
  57                  tmp_working_dir,
  58                  testbase="",
  59                  grids=None,
  60                  sessions=None,
  61                  launcher="",
  62                  show_desktop=True):
  63         self.grids = grids
  64         self.config = config
  65         self.logger = logger
  66         self.tmp_working_dir = tmp_working_dir
  67         self.sessions = sessions
  68         self.launcher = launcher
  69         self.show_desktop = show_desktop
  70
  71         res = self.prepare_testbase(testbase)
  72         self.test_base_found = True
  73         if res == 1:
  74             # Fail
  75             self.test_base_found = False
  76
  77         self.settings = {}
  78         self.known_errors = None
  79
  80         # create section for results
  81         self.config.TESTS = src.pyconf.Sequence(self.config)
  82
  83         self.nb_run = 0
  84         self.nb_succeed = 0
  85         self.nb_timeout = 0
  86         self.nb_not_run = 0
  87         self.nb_acknoledge = 0
  88
  89     def _copy_dir(self, source, target):
  90         if self.config.VARS.python >= "2.6":
  91             shutil.copytree(source, target,
  92                             symlinks=True,
  93                             ignore=shutil.ignore_patterns('.git*','.svn*'))
  94         else:
  95             shutil.copytree(source, target,
  96                             symlinks=True)
  97
  98     def prepare_testbase_from_dir(self, testbase_name, testbase_dir):
  99         self.logger.write(_("get test base from dir: %s\n") % \
 100                           src.printcolors.printcLabel(testbase_dir), 3)
 101         if not os.access(testbase_dir, os.X_OK):
 102             raise src.SatException(_("testbase %(name)s (%(dir)s) does not "
 103                                      "exist ...\n") % { 'name': testbase_name,
 104                                                        'dir': testbase_dir })
 105
 106         self._copy_dir(testbase_dir,
 107                        os.path.join(self.tmp_working_dir, 'BASES', testbase_name))
 108
 109     def prepare_testbase_from_git(self,
 110                                   testbase_name,
 111                                   testbase_base,
 112                                   testbase_tag):
 113         self.logger.write(
 114             _("get test base '%(testbase)s' with '%(tag)s' tag from git\n") % {
 115                         "testbase" : src.printcolors.printcLabel(testbase_name),
 116                         "tag" : src.printcolors.printcLabel(testbase_tag)},
 117                           3)
 118         try:
 119             def set_signal(): # pragma: no cover
 120                 """see http://bugs.python.org/issue1652"""
 121                 import signal
 122                 signal.signal(signal.SIGPIPE, signal.SIG_DFL)
 123
 124             cmd = "git clone --depth 1 %(base)s %(dir)s"
 125             cmd += " && cd %(dir)s"
 126             if testbase_tag=='master':
 127                 cmd += " && git fetch origin %(branch)s"
 128             else:
 129                 cmd += " && git fetch origin %(branch)s:%(branch)s"
 130             cmd += " && git checkout %(branch)s"
 131             cmd = cmd % { 'branch': testbase_tag,
 132                          'base': testbase_base,
 133                          'dir': testbase_name }
 134
 135             self.logger.write("> %s\n" % cmd, 5)
 136             if src.architecture.is_windows():
 137                 # preexec_fn not supported on windows platform
 138                 res = subprocess.call(cmd,
 139                                 cwd=os.path.join(self.tmp_working_dir, 'BASES'),
 140                                 shell=True,
 141                                 stdout=self.logger.logTxtFile,
 142                                 stderr=subprocess.PIPE)
 143             else:
 144                 res = subprocess.call(cmd,
 145                                 cwd=os.path.join(self.tmp_working_dir, 'BASES'),
 146                                 shell=True,
 147                                 preexec_fn=set_signal,
 148                                 stdout=self.logger.logTxtFile,
 149                                 stderr=subprocess.PIPE)
 150             if res != 0:
 151                 raise src.SatException(_("Error: unable to get test base "
 152                                          "'%(name)s' from git '%(repo)s'.") % \
 153                                        { 'name': testbase_name,
 154                                         'repo': testbase_base })
 155
 156         except OSError:
 157             self.logger.error(_("git is not installed. exiting...\n"))
 158             sys.exit(0)
 159
 160     def prepare_testbase_from_svn(self, user, testbase_name, testbase_base):
 161         self.logger.write(_("get test base '%s' from svn\n") % \
 162                           src.printcolors.printcLabel(testbase_name), 3)
 163         try:
 164             def set_signal(): # pragma: no cover
 165                 """see http://bugs.python.org/issue1652"""
 166                 import signal
 167                 signal.signal(signal.SIGPIPE, signal.SIG_DFL)
 168
 169             cmd = "svn checkout --username %(user)s %(base)s %(dir)s"
 170             cmd = cmd % { 'user': user,
 171                          'base': testbase_base,
 172                          'dir': testbase_name }
 173
 174             self.logger.write("> %s\n" % cmd, 5)
 175             if src.architecture.is_windows():
 176                 # preexec_fn not supported on windows platform
 177                 res = subprocess.call(cmd,
 178                                 cwd=os.path.join(self.tmp_working_dir, 'BASES'),
 179                                 shell=True,
 180                                 stdout=self.logger.logTxtFile,
 181                                 stderr=subprocess.PIPE)
 182             else:
 183                 res = subprocess.call(cmd,
 184                                 cwd=os.path.join(self.tmp_working_dir, 'BASES'),
 185                                 shell=True,
 186                                 preexec_fn=set_signal,
 187                                 stdout=self.logger.logTxtFile,
 188                                 stderr=subprocess.PIPE)
 189
 190             if res != 0:
 191                 raise src.SatException(_("Error: unable to get test base '%(nam"
 192                                          "e)s' from svn '%(repo)s'.") % \
 193                                        { 'name': testbase_name,
 194                                         'repo': testbase_base })
 195
 196         except OSError:
 197             self.logger.error(_("svn is not installed. exiting...\n"))
 198             sys.exit(0)
 199
 200     ##
 201     # Configure tests base.
 202     def prepare_testbase(self, test_base_name):
 203         src.printcolors.print_value(self.logger,
 204                                     _("Test base"),
 205                                     test_base_name,
 206                                     3)
 207         self.logger.write("\n", 3, False)
 208
 209         # search for the test base
 210         test_base_info = None
 211         for project_name in self.config.PROJECTS.projects:
 212             project_info = self.config.PROJECTS.projects[project_name]
 213             for t_b_info in project_info.test_bases:
 214                 if t_b_info.name == test_base_name:
 215                     test_base_info = t_b_info
 216
 217         if not test_base_info:
 218             if os.path.exists(test_base_name):
 219                 self.prepare_testbase_from_dir("DIR", test_base_name)
 220                 self.currentTestBase = "DIR"
 221                 return 0
 222
 223         if not test_base_info:
 224             message = (_("########## ERROR: test base '%s' not found\n") %
 225                        test_base_name)
 226             self.logger.write("%s\n" % src.printcolors.printcError(message))
 227             return 1
 228
 229         if test_base_info.get_sources == "dir":
 230             self.prepare_testbase_from_dir(test_base_name,
 231                                            test_base_info.info.dir)
 232         elif test_base_info.get_sources == "git":
 233             self.prepare_testbase_from_git(test_base_name,
 234                                        test_base_info.info.base,
 235                                        self.config.APPLICATION.test_base.tag)
 236         elif test_base_info.get_sources == "svn":
 237             svn_user = src.get_cfg_param(test_base_info.svn_info,
 238                                          "svn_user",
 239                                          self.config.USER.svn_user)
 240             self.prepare_testbase_from_svn(svn_user,
 241                                        test_base_name,
 242                                        test_base_info.info.base)
 243         else:
 244             raise src.SatException(_("unknown source type '%(type)s' for test b"
 245                                      "ase '%(base)s' ...\n") % {
 246                                         'type': test_base_info.get_sources,
 247                                         'base': test_base_name })
 248
 249         self.currentTestBase = test_base_name
 250
 251     ##
 252     # Searches if the script is declared in known errors pyconf.
 253     # Update the status if needed.
 254     def search_known_errors(self, status, test_grid, test_session, test):
 255         test_path = os.path.join(test_grid, test_session, test)
 256         if not src.config_has_application(self.config):
 257             return status, []
 258
 259         if self.known_errors is None:
 260             return status, []
 261
 262         platform = self.config.VARS.arch
 263         application = self.config.VARS.application
 264         error = self.known_errors.get_error(test_path, application, platform)
 265         if error is None:
 266             return status, []
 267
 268         if status == src.OK_STATUS:
 269             if not error.fixed:
 270                 # the error is fixed
 271                 self.known_errors.fix_error(error)
 272                 #import testerror
 273                 #testerror.write_test_failures(
 274                 #                        self.config.TOOLS.testerror.file_path,
 275                 #                        self.known_errors.errors)
 276             return status, [ error.date,
 277                             error.expected,
 278                             error.comment,
 279                             error.fixed ]
 280
 281         if error.fixed:
 282             self.known_errors.unfix_error(error)
 283             #import testerror
 284             #testerror.write_test_failures(self.config.TOOLS.testerror.file_path,
 285             #                              self.known_errors.errors)
 286
 287         delta = self.known_errors.get_expecting_days(error)
 288         kfres = [ error.date, error.expected, error.comment, error.fixed ]
 289         if delta < 0:
 290             return src.KO_STATUS, kfres
 291         return src.KNOWNFAILURE_STATUS, kfres
 292
 293     ##
 294     # Read the *.result.py files.
 295     def read_results(self, listTest, has_timed_out):
 296         results = {}
 297         for test in listTest:
 298             resfile = os.path.join(self.currentDir,
 299                                    self.currentgrid,
 300                                    self.currentsession,
 301                                    test[:-3] + ".result.py")
 302
 303             # check if <test>.result.py file exists
 304             if not os.path.exists(resfile):
 305                 results[test] = ["?", -1, "", []]
 306             else:
 307                 gdic, ldic = {}, {}
 308                 execfile(resfile, gdic, ldic)
 309
 310                 status = src.TIMEOUT_STATUS
 311                 if not has_timed_out:
 312                     status = src.KO_STATUS
 313
 314                 if ldic.has_key('status'):
 315                     status = ldic['status']
 316
 317                 expected = []
 318                 if status == src.KO_STATUS or status == src.OK_STATUS:
 319                     status, expected = self.search_known_errors(status,
 320                                                             self.currentgrid,
 321                                                             self.currentsession,
 322                                                             test)
 323
 324                 callback = ""
 325                 if ldic.has_key('callback'):
 326                     callback = ldic['callback']
 327                 elif status == src.KO_STATUS:
 328                     callback = "CRASH"
 329
 330                 exec_time = -1
 331                 if ldic.has_key('time'):
 332                     try:
 333                         exec_time = float(ldic['time'])
 334                     except:
 335                         pass
 336
 337                 results[test] = [status, exec_time, callback, expected]
 338
 339             # check if <test>.py file exists
 340             testfile = os.path.join(self.currentDir,
 341                                    self.currentgrid,
 342                                    self.currentsession,
 343                                    test)
 344
 345             if not os.path.exists(testfile):
 346                 results[test].append('')
 347             else:
 348                 text = open(testfile, "r").read()
 349                 results[test].append(text)
 350
 351             # check if <test>.out.py file exists
 352             outfile = os.path.join(self.currentDir,
 353                                    self.currentgrid,
 354                                    self.currentsession,
 355                                    test[:-3] + ".out.py")
 356
 357             if not os.path.exists(outfile):
 358                 results[test].append('')
 359             else:
 360                 text = open(outfile, "r").read()
 361                 results[test].append(text)
 362
 363         return results
 364
 365     ##
 366     # Generates the script to be run by Salome.
 367     # This python script includes init and close statements and a loop
 368     # calling all the scripts of a single directory.
 369     def generate_script(self, listTest, script_path, ignoreList):
 370         # open template file
 371         template_file = open(os.path.join(self.config.VARS.srcDir,
 372                                           "test",
 373                                           "scriptTemplate.py"), 'r')
 374         template = string.Template(template_file.read())
 375
 376         # create substitution dictionary
 377         d = dict()
 378         d['resourcesWay'] = os.path.join(self.currentDir, 'RESSOURCES')
 379         d['tmpDir'] = os.path.join(self.tmp_working_dir, 'WORK')
 380         d['toolsWay'] = os.path.join(self.config.VARS.srcDir, "test")
 381         d['sessionDir'] = os.path.join(self.currentDir,
 382                                     self.currentgrid,
 383                                     self.currentsession)
 384         d['resultFile'] = os.path.join(self.tmp_working_dir,
 385                                        'WORK',
 386                                        'exec_result')
 387         d['listTest'] = listTest
 388         d['sessionName'] = self.currentsession
 389         d['ignore'] = ignoreList
 390
 391         # create script with template
 392         script = open(script_path, 'w')
 393         script.write(template.safe_substitute(d))
 394         script.close()
 395
 396     # Find the getTmpDir function that gives access to *pidict file directory.
 397     # (the *pidict file exists when SALOME is launched)
 398     def get_tmp_dir(self):
 399         # Rare case where there is no KERNEL in grid list
 400         # (for example MED_STANDALONE)
 401         if ('APPLICATION' in self.config
 402                 and 'KERNEL' not in self.config.APPLICATION.products
 403                 and 'KERNEL_ROOT_DIR' not in os.environ):
 404             return getTmpDirDEFAULT
 405
 406         # Case where "sat test" is launched in an existing SALOME environment
 407         if 'KERNEL_ROOT_DIR' in os.environ:
 408             root_dir =  os.environ['KERNEL_ROOT_DIR']
 409
 410         if ('APPLICATION' in self.config
 411                 and 'KERNEL' in self.config.APPLICATION.products):
 412             root_dir = src.product.get_product_config(self.config,
 413                                                       "KERNEL").install_dir
 414
 415         # Case where there the appli option is called (with path to launcher)
 416         if len(self.launcher) > 0:
 417             # There are two cases : The old application (runAppli)
 418             # and the new one
 419             launcherName = os.path.basename(self.launcher)
 420             launcherDir = os.path.dirname(self.launcher)
 421             if launcherName == 'runAppli':
 422                 # Old application
 423                 cmd = "for i in " + launcherDir + "/env.d/*.sh; do source ${i};"
 424                 " done ; echo $KERNEL_ROOT_DIR"
 425             else:
 426                 # New application
 427                 cmd = "echo -e 'import os\nprint os.environ[\"KERNEL_ROOT_DIR\""
 428                 "]' > tmpscript.py; %s shell tmpscript.py" % self.launcher
 429             root_dir = subprocess.Popen(cmd,
 430                             stdout=subprocess.PIPE,
 431                             shell=True,
 432                             executable='/bin/bash').communicate()[0].split()[-1]
 433
 434         # import grid salome_utils from KERNEL that gives
 435         # the right getTmpDir function
 436
 437         (file_, pathname, description) = imp.find_module("salome_utils",
 438                                                          [os.path.join(root_dir,
 439                                                                     'bin',
 440                                                                     'salome')])
 441         try:
 442             grid = imp.load_module("salome_utils",
 443                                      file_,
 444                                      pathname,
 445                                      description)
 446             return grid.getLogDir
 447         except:
 448             grid = imp.load_module("salome_utils",
 449                                      file_,
 450                                      pathname,
 451                                      description)
 452             return grid.getTmpDir
 453         finally:
 454             if file_:
 455                 file_.close()
 456
 457
 458     def get_test_timeout(self, test_name, default_value):
 459         if ("timeout" in self.settings and
 460                 test_name in self.settings["timeout"]):
 461             return self.settings["timeout"][test_name]
 462
 463         return default_value
 464
 465     def generate_launching_commands(self):
 466         # Case where "sat test" is launched in an existing SALOME environment
 467         if 'KERNEL_ROOT_DIR' in os.environ:
 468             binSalome = "runSalome"
 469             binPython = "python"
 470             killSalome = "killSalome.py"
 471
 472         # Rare case where there is no KERNEL in grid list
 473         # (for example MED_STANDALONE)
 474         if ('APPLICATION' in self.config and
 475                 'KERNEL' not in self.config.APPLICATION.products):
 476             binSalome = "runSalome"
 477             binPython = "python"
 478             killSalome = "killSalome.py"
 479             src.environment.load_environment(self.config, False, self.logger)
 480             return binSalome, binPython, killSalome
 481
 482         # Case where there the appli option is called (with path to launcher)
 483         if len(self.launcher) > 0:
 484             # There are two cases : The old application (runAppli)
 485             # and the new one
 486             launcherName = os.path.basename(self.launcher)
 487             launcherDir = os.path.dirname(self.launcher)
 488             if launcherName == 'runAppli':
 489                 # Old application
 490                 binSalome = self.launcher
 491                 binPython = ("for i in " +
 492                              launcherDir +
 493                              "/env.d/*.sh; do source ${i}; done ; python")
 494                 killSalome = ("for i in " +
 495                         launcherDir +
 496                         "/env.d/*.sh; do source ${i}; done ; killSalome.py'")
 497                 return binSalome, binPython, killSalome
 498             else:
 499                 # New application
 500                 binSalome = self.launcher
 501                 binPython = self.launcher + ' shell'
 502                 killSalome = self.launcher + ' killall'
 503                 return binSalome, binPython, killSalome
 504
 505         # SALOME version detection and APPLI repository detection
 506         VersionSalome = src.get_salome_version(self.config)
 507         appdir = 'APPLI'
 508         if "APPLI" in self.config and "application_name" in self.config.APPLI:
 509             appdir = self.config.APPLI.application_name
 510
 511         # Case where SALOME has NOT the launcher that uses the SalomeContext API
 512         if VersionSalome < 730:
 513             binSalome = os.path.join(self.config.APPLI.grid_appli_install_dir,
 514                                      appdir,
 515                                      "runAppli")
 516             binPython = "python"
 517             killSalome = "killSalome.py"
 518             src.environment.load_environment(self.config, False, self.logger)
 519             return binSalome, binPython, killSalome
 520
 521         # Case where SALOME has the launcher that uses the SalomeContext API
 522         if VersionSalome >= 730:
 523             if 'profile' not in self.config.APPLICATION:
 524                 # Before revision of application concept
 525                 launcher_name = self.config.APPLI.launch_alias_name
 526                 binSalome = os.path.join(self.config.APPLICATION.workdir,
 527                                          appdir,
 528                                          launcher_name)
 529             else:
 530                 # After revision of application concept
 531                 launcher_name = self.config.APPLICATION.profile.launcher_name
 532                 binSalome = os.path.join(self.config.APPLICATION.workdir,
 533                                          launcher_name)
 534
 535             if src.architecture.is_windows():
 536                 binSalome += '.bat'
 537
 538             binPython = binSalome + ' shell'
 539             killSalome = binSalome + ' killall'
 540             return binSalome, binPython, killSalome
 541
 542         return binSalome, binPython, killSalome
 543
 544
 545     ##
 546     # Runs tests of a session (using a single instance of Salome).
 547     def run_tests(self, listTest, ignoreList):
 548         out_path = os.path.join(self.currentDir,
 549                                 self.currentgrid,
 550                                 self.currentsession)
 551         sessionname = "%s/%s" % (self.currentgrid, self.currentsession)
 552         time_out = self.get_test_timeout(sessionname,
 553                                          self.config.SITE.test.timeout)
 554
 555         time_out_salome = src.get_cfg_param(self.config.SITE.test,
 556                                             "timeout_app",
 557                                             self.config.SITE.test.timeout)
 558
 559         # generate wrapper script
 560         script_path = os.path.join(out_path, 'wrapperScript.py')
 561         self.generate_script(listTest, script_path, ignoreList)
 562
 563         tmpDir = self.get_tmp_dir()
 564
 565         binSalome, binPython, killSalome = self.generate_launching_commands()
 566         if self.settings.has_key("run_with_grids") \
 567            and self.settings["run_with_grids"].has_key(sessionname):
 568             binSalome = (binSalome +
 569                          " -m %s" % self.settings["run_with_grids"][sessionname])
 570
 571         logWay = os.path.join(self.tmp_working_dir, "WORK", "log_cxx")
 572
 573         status = False
 574         elapsed = -1
 575         if self.currentsession.startswith("NOGUI_"):
 576             # runSalome -t (bash)
 577             status, elapsed = fork.batch(binSalome, self.logger,
 578                                         os.path.join(self.tmp_working_dir,
 579                                                      "WORK"),
 580                                         [ "-t",
 581                                          "--shutdown-server=1",
 582                                          script_path ],
 583                                         delai=time_out,
 584                                         log=logWay)
 585
 586         elif self.currentsession.startswith("PY_"):
 587             # python script.py
 588             status, elapsed = fork.batch(binPython, self.logger,
 589                                           os.path.join(self.tmp_working_dir,
 590                                                        "WORK"),
 591                                           [script_path],
 592                                           delai=time_out, log=logWay)
 593
 594         else:
 595             opt = "-z 0"
 596             if self.show_desktop: opt = "--show-desktop=0"
 597             status, elapsed = fork.batch_salome(binSalome,
 598                                                  self.logger,
 599                                                  os.path.join(
 600                                                         self.tmp_working_dir,
 601                                                         "WORK"),
 602                                                  [ opt,
 603                                                   "--shutdown-server=1",
 604                                                   script_path ],
 605                                                  getTmpDir=tmpDir,
 606                                                  fin=killSalome,
 607                                                  delai=time_out,
 608                                                  log=logWay,
 609                                                  delaiapp=time_out_salome)
 610
 611         self.logger.write("status = %s, elapsed = %s\n" % (status, elapsed),
 612                           5)
 613
 614         # create the test result to add in the config object
 615         test_info = src.pyconf.Mapping(self.config)
 616         test_info.testbase = self.currentTestBase
 617         test_info.grid = self.currentgrid
 618         test_info.session = self.currentsession
 619         test_info.script = src.pyconf.Sequence(self.config)
 620
 621         script_results = self.read_results(listTest, elapsed == time_out)
 622         for sr in sorted(script_results.keys()):
 623             self.nb_run += 1
 624
 625             # create script result
 626             script_info = src.pyconf.Mapping(self.config)
 627             script_info.name = sr
 628             script_info.res = script_results[sr][0]
 629             script_info.time = script_results[sr][1]
 630             if script_info.res == src.TIMEOUT_STATUS:
 631                 script_info.time = time_out
 632             if script_info.time < 1e-3: script_info.time = 0
 633
 634             callback = script_results[sr][2]
 635             if script_info.res != src.OK_STATUS and len(callback) > 0:
 636                 script_info.callback = callback
 637
 638             kfres = script_results[sr][3]
 639             if len(kfres) > 0:
 640                 script_info.known_error = src.pyconf.Mapping(self.config)
 641                 script_info.known_error.date = kfres[0]
 642                 script_info.known_error.expected = kfres[1]
 643                 script_info.known_error.comment = kfres[2]
 644                 script_info.known_error.fixed = kfres[3]
 645
 646             script_info.content = script_results[sr][4]
 647             script_info.out = script_results[sr][5]
 648
 649             # add it to the list of results
 650             test_info.script.append(script_info, '')
 651
 652             # display the results
 653             if script_info.time > 0:
 654                 exectime = "(%7.3f s)" % script_info.time
 655             else:
 656                 exectime = ""
 657
 658             sp = "." * (35 - len(script_info.name))
 659             self.logger.write(self.write_test_margin(3), 3)
 660             self.logger.write("script %s %s %s %s\n" % (
 661                                 src.printcolors.printcLabel(script_info.name),
 662                                 sp,
 663                                 src.printcolors.printc(script_info.res),
 664                                 exectime), 3, False)
 665             if script_info and len(callback) > 0:
 666                 self.logger.write("Exception in %s\n%s\n" % \
 667                     (script_info.name,
 668                      src.printcolors.printcWarning(callback)), 2, False)
 669
 670             if script_info.res == src.OK_STATUS:
 671                 self.nb_succeed += 1
 672             elif script_info.res == src.KNOWNFAILURE_STATUS:
 673                 self.nb_acknoledge += 1
 674             elif script_info.res == src.TIMEOUT_STATUS:
 675                 self.nb_timeout += 1
 676             elif script_info.res == src.NA_STATUS:
 677                 self.nb_run -= 1
 678             elif script_info.res == "?":
 679                 self.nb_not_run += 1
 680
 681         self.config.TESTS.append(test_info, '')
 682
 683     ##
 684     # Runs all tests of a session.
 685     def run_session_tests(self):
 686
 687         self.logger.write(self.write_test_margin(2), 3)
 688         self.logger.write("Session = %s\n" % src.printcolors.printcLabel(
 689                                                     self.currentsession), 3, False)
 690
 691         # prepare list of tests to run
 692         tests = os.listdir(os.path.join(self.currentDir,
 693                                         self.currentgrid,
 694                                         self.currentsession))
 695         tests = filter(lambda l: l.endswith(".py"), tests)
 696         tests = sorted(tests, key=str.lower)
 697
 698         # build list of known failures
 699         cat = "%s/%s/" % (self.currentgrid, self.currentsession)
 700         ignoreDict = {}
 701         for k in self.ignore_tests.keys():
 702             if k.startswith(cat):
 703                 ignoreDict[k[len(cat):]] = self.ignore_tests[k]
 704
 705         self.run_tests(tests, ignoreDict)
 706
 707     ##
 708     # Runs all tests of a grid.
 709     def run_grid_tests(self):
 710         self.logger.write(self.write_test_margin(1), 3)
 711         self.logger.write("grid = %s\n" % src.printcolors.printcLabel(
 712                                                 self.currentgrid), 3, False)
 713
 714         grid_path = os.path.join(self.currentDir, self.currentgrid)
 715
 716         sessions = []
 717         if self.sessions is not None:
 718             sessions = self.sessions # user choice
 719         else:
 720             # use all scripts in grid
 721             sessions = filter(lambda l: l not in C_IGNORE_GRIDS,
 722                            os.listdir(grid_path))
 723             sessions = filter(lambda l: os.path.isdir(os.path.join(grid_path,
 724                                                                 l)), sessions)
 725
 726         sessions = sorted(sessions, key=str.lower)
 727         for session_ in sessions:
 728             if not os.path.exists(os.path.join(grid_path, session_)):
 729                 self.logger.write(self.write_test_margin(2), 3)
 730                 self.logger.write(src.printcolors.printcWarning("Session %s not"
 731                                         " found" % session_) + "\n", 3, False)
 732             else:
 733                 self.currentsession = session_
 734                 self.run_session_tests()
 735
 736     ##
 737     # Runs test testbase.
 738     def run_testbase_tests(self):
 739         res_dir = os.path.join(self.currentDir, "RESSOURCES")
 740         os.environ['PYTHONPATH'] =  (res_dir +
 741                                      os.pathsep +
 742                                      os.environ['PYTHONPATH'])
 743         os.environ['TT_BASE_RESSOURCES'] = res_dir
 744         src.printcolors.print_value(self.logger,
 745                                     "TT_BASE_RESSOURCES",
 746                                     res_dir,
 747                                     4)
 748         self.logger.write("\n", 4, False)
 749
 750         self.logger.write(self.write_test_margin(0), 3)
 751         testbase_label = "Test base = %s\n" % src.printcolors.printcLabel(
 752                                                         self.currentTestBase)
 753         self.logger.write(testbase_label, 3, False)
 754         self.logger.write("-" * len(src.printcolors.cleancolor(testbase_label)),
 755                           3)
 756         self.logger.write("\n", 3, False)
 757
 758         # load settings
 759         settings_file = os.path.join(res_dir, "test_settings.py")
 760         if os.path.exists(settings_file):
 761             gdic, ldic = {}, {}
 762             execfile(settings_file, gdic, ldic)
 763             self.logger.write(_("Load test settings\n"), 3)
 764             self.settings = ldic['settings_dic']
 765             self.ignore_tests = ldic['known_failures_list']
 766             if isinstance(self.ignore_tests, list):
 767                 self.ignore_tests = {}
 768                 self.logger.write(src.printcolors.printcWarning("known_failur"
 769                   "es_list must be a dictionary (not a list)") + "\n", 1, False)
 770         else:
 771             self.ignore_tests = {}
 772             self.settings.clear()
 773
 774         # read known failures pyconf
 775         if "testerror" in self.config.SITE:
 776             #import testerror
 777             #self.known_errors = testerror.read_test_failures(
 778             #                            self.config.TOOLS.testerror.file_path,
 779             #                            do_error=False)
 780             pass
 781         else:
 782             self.known_errors = None
 783
 784         if self.grids is not None:
 785             grids = self.grids # given by user
 786         else:
 787             # select all the grids (i.e. directories) in the directory
 788             grids = filter(lambda l: l not in C_IGNORE_GRIDS,
 789                              os.listdir(self.currentDir))
 790             grids = filter(lambda l: os.path.isdir(
 791                                         os.path.join(self.currentDir, l)),
 792                              grids)
 793
 794         grids = sorted(grids, key=str.lower)
 795         for grid in grids:
 796             if not os.path.exists(os.path.join(self.currentDir, grid)):
 797                 self.logger.write(self.write_test_margin(1), 3)
 798                 self.logger.write(src.printcolors.printcWarning(
 799                             "grid %s does not exist\n" % grid), 3, False)
 800             else:
 801                 self.currentgrid = grid
 802                 self.run_grid_tests()
 803
 804     def run_script(self, script_name):
 805         if ('APPLICATION' in self.config and
 806                 script_name in self.config.APPLICATION):
 807             script = self.config.APPLICATION[script_name]
 808             if len(script) == 0:
 809                 return
 810
 811             self.logger.write("\n", 2, False)
 812             if not os.path.exists(script):
 813                 self.logger.write(src.printcolors.printcWarning("WARNING: scrip"
 814                                         "t not found: %s" % script) + "\n", 2)
 815             else:
 816                 self.logger.write(src.printcolors.printcHeader("----------- sta"
 817                                             "rt %s" % script_name) + "\n", 2)
 818                 self.logger.write("Run script: %s\n" % script, 2)
 819                 subprocess.Popen(script, shell=True).wait()
 820                 self.logger.write(src.printcolors.printcHeader("----------- end"
 821                                                 " %s" % script_name) + "\n", 2)
 822
 823     def run_all_tests(self):
 824         initTime = datetime.datetime.now()
 825
 826         self.run_script('test_setup')
 827         self.logger.write("\n", 2, False)
 828
 829         self.logger.write(src.printcolors.printcHeader(
 830                                             _("=== STARTING TESTS")) + "\n", 2)
 831         self.logger.write("\n", 2, False)
 832         self.currentDir = os.path.join(self.tmp_working_dir,
 833                                        'BASES',
 834                                        self.currentTestBase)
 835         self.run_testbase_tests()
 836
 837         # calculate total execution time
 838         totalTime = datetime.datetime.now() - initTime
 839         totalTime -= datetime.timedelta(microseconds=totalTime.microseconds)
 840         self.logger.write("\n", 2, False)
 841         self.logger.write(src.printcolors.printcHeader(_("=== END TESTS")), 2)
 842         self.logger.write(" %s\n" % src.printcolors.printcInfo(str(totalTime)),
 843                           2,
 844                           False)
 845
 846         #
 847         # Start the tests
 848         #
 849         self.run_script('test_cleanup')
 850         self.logger.write("\n", 2, False)
 851
 852         # evaluate results
 853         res_count = "%d / %d" % (self.nb_succeed,
 854                                  self.nb_run - self.nb_acknoledge)
 855
 856         res_out = _("Tests Results: %(succeed)d / %(total)d\n") % \
 857             { 'succeed': self.nb_succeed, 'total': self.nb_run }
 858         if self.nb_succeed == self.nb_run:
 859             res_out = src.printcolors.printcSuccess(res_out)
 860         else:
 861             res_out = src.printcolors.printcError(res_out)
 862         self.logger.write(res_out, 1)
 863
 864         if self.nb_timeout > 0:
 865             self.logger.write(_("%d tests TIMEOUT\n") % self.nb_timeout, 1)
 866             res_count += " TO: %d" % self.nb_timeout
 867         if self.nb_not_run > 0:
 868             self.logger.write(_("%d tests not executed\n") % self.nb_not_run, 1)
 869             res_count += " NR: %d" % self.nb_not_run
 870
 871         status = src.OK_STATUS
 872         if self.nb_run - self.nb_succeed - self.nb_acknoledge > 0:
 873             status = src.KO_STATUS
 874         elif self.nb_acknoledge:
 875             status = src.KNOWNFAILURE_STATUS
 876
 877         self.logger.write(_("Status: %s\n" % status), 3)
 878
 879         return self.nb_run - self.nb_succeed - self.nb_acknoledge
 880
 881     ##
 882     # Write margin to show test results.
 883     def write_test_margin(self, tab):
 884         if tab == 0:
 885             return ""
 886         return "|   " * (tab - 1) + "+ "
 887