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