Merge remote branch 'autotest-upstream/master' into autotest-merge

client/bin/job.py

 """The main job wrapper
 
 This is the core infrastructure.
 
 Copyright Andy Whitcroft, Martin J. Bligh 2006
 """
 
 import copy, os, platform, re, shutil, sys, time, traceback, types, glob
 import logging, getpass, errno, weakref
 import cPickle as pickle
@@ skipping 175 matching lines @@
 
         # set up the status logger
         def client_job_record_hook(entry):
             msg_tag = ''
             if '.' in self._logger.global_filename:
                 msg_tag = self._logger.global_filename.split('.', 1)[1]
             # send the entry to the job harness
             message = '\n'.join([entry.message] + entry.extra_message_lines)
             rendered_entry = self._logger.render_entry(entry)
             self.harness.test_status_detail(entry.status_code, entry.subdir,
-                                            entry.operation, message, msg_tag)
+                                            entry.operation, message, msg_tag,
+                                            entry.fields)
             self.harness.test_status(rendered_entry, msg_tag)
             # send the entry to stdout, if it's enabled
             logging.info(rendered_entry)
         self._logger = base_job.status_logger(
             self, status_indenter(self), record_hook=client_job_record_hook,
             tap_writer=self._tap)
 
     def _post_record_init(self, control, options, drop_caches,
                           extra_copy_cmdline):
         """
@@ skipping 301 matching lines @@
 
             # dep_dir might not exist if it is not fetched from the repos
             if not os.path.exists(dep_dir):
                 raise error.TestError("Dependency %s does not exist" % dep)
 
             os.chdir(dep_dir)
             if execfile('%s.py' % dep, {}) is None:
                 logging.info('Dependency %s successfuly built', dep)
 
 
-    def _runtest(self, url, tag, args, dargs):
+    def _runtest(self, url, tag, timeout, args, dargs):
         try:
             l = lambda : test.runtest(self, url, tag, args, dargs)
             pid = parallel.fork_start(self.resultdir, l)
-            parallel.fork_waitfor(self.resultdir, pid)
+
+            if timeout:
+                logging.debug('Waiting for pid %d for %d seconds', pid, timeout)
+                parallel.fork_waitfor_timed(self.resultdir, pid, timeout)
+            else:
+                parallel.fork_waitfor(self.resultdir, pid)
+
         except error.TestBaseException:
             # These are already classified with an error type (exit_status)
             raise
         except error.JobError:
             raise  # Caught further up and turned into an ABORT.
         except Exception, e:
             # Converts all other exceptions thrown by the test regardless
             # of phase into a TestError(TestBaseException) subclass that
             # reports them with their full stack trace.
             raise error.UnhandledTestError(e)
 
 
-    @_run_test_complete_on_exit
-    def run_test(self, url, *args, **dargs):
+    def _run_test_base(self, url, *args, **dargs):
         """
-        Summon a test object and run it.
+        Prepares arguments and run functions to run_test and run_test_detail.
 
         @param url A url that identifies the test to run.
         @param tag An optional keyword argument that will be added to the
             test and subdir name.
         @param subdir_tag An optional keyword argument that will be added
             to the subdir name.
 
-        @returns True if the test passes, False otherwise.
+        @returns:
+                subdir: Test subdirectory
+                testname: Test name
+                group_func: Actual test run function
+                timeout: Test timeout
         """
         group, testname = self.pkgmgr.get_package_name(url, 'test')
         testname, subdir, tag = self._build_tagged_test_name(testname, dargs)
         outputdir = self._make_test_outputdir(subdir)
 
+        timeout = dargs.pop('timeout', None)
+        if timeout:
+            logging.debug('Test has timeout: %d sec.', timeout)
+
         def log_warning(reason):
             self.record("WARN", subdir, testname, reason)
         @disk_usage_monitor.watch(log_warning, "/", self._max_disk_usage_rate)
         def group_func():
             try:
-                self._runtest(url, tag, args, dargs)
+                self._runtest(url, tag, timeout, args, dargs)
             except error.TestBaseException, detail:
                 # The error is already classified, record it properly.
                 self.record(detail.exit_status, subdir, testname, str(detail))
                 raise
             else:
                 self.record('GOOD', subdir, testname, 'completed successfully')
 
+        return (subdir, testname, group_func, timeout)
+
+
+    @_run_test_complete_on_exit
+    def run_test(self, url, *args, **dargs):
+        """
+        Summon a test object and run it.
+
+        @param url A url that identifies the test to run.
+        @param tag An optional keyword argument that will be added to the
+            test and subdir name.
+        @param subdir_tag An optional keyword argument that will be added
+            to the subdir name.
+
+        @returns True if the test passes, False otherwise.
+        """
+        (subdir, testname, group_func, timeout) = self._run_test_base(url,
+                                                                      *args,
+                                                                      **dargs)
         try:
-            self._rungroup(subdir, testname, group_func)
+            self._rungroup(subdir, testname, group_func, timeout)
             return True
         except error.TestBaseException:
             return False
         # Any other exception here will be given to the caller
         #
         # NOTE: The only exception possible from the control file here
         # is error.JobError as _runtest() turns all others into an
         # UnhandledTestError that is caught above.
 
 
-    def _rungroup(self, subdir, testname, function, *args, **dargs):
+    @_run_test_complete_on_exit
+    def run_test_detail(self, url, *args, **dargs):
+        """
+        Summon a test object and run it, returning test status.
+
+        @param url A url that identifies the test to run.
+        @param tag An optional keyword argument that will be added to the
+            test and subdir name.
+        @param subdir_tag An optional keyword argument that will be added
+            to the subdir name.
+
+        @returns Test status
+        @see: client/common_lib/error.py, exit_status
+        """
+        (subdir, testname, group_func, timeout) = self._run_test_base(url,
+                                                                      *args,
+                                                                      **dargs)
+        try:
+            self._rungroup(subdir, testname, group_func, timeout)
+            return 'GOOD'
+        except error.TestBaseException, detail:
+            return detail.exit_status
+
+
+    def _rungroup(self, subdir, testname, function, timeout, *args, **dargs):
         """\
         subdir:
                 name of the group
         testname:
                 name of the test to run, or support step
         function:
                 subroutine to run
         *args:
                 arguments for the function
 
         Returns the result of the passed in function
         """
 
         try:
-            self.record('START', subdir, testname)
+            optional_fields = None
+            if timeout:
+                optional_fields = {}
+                optional_fields['timeout'] = timeout
+            self.record('START', subdir, testname,
+                        optional_fields=optional_fields)
+
             self._state.set('client', 'unexpected_reboot', (subdir, testname))
             try:
                 result = function(*args, **dargs)
                 self.record('END GOOD', subdir, testname)
                 return result
             except error.TestBaseException, e:
                 self.record('END %s' % e.exit_status, subdir, testname)
                 raise
             except error.JobError, e:
                 self.record('END ABORT', subdir, testname)
@@ skipping 23 matching lines @@
         **dargs:
                 Named arguments for the function.
         """
         if tag:
             name = tag
         else:
             name = function.__name__
 
         try:
             return self._rungroup(subdir=None, testname=name,
-                                  function=function, **dargs)
+                                  function=function, timeout=None, **dargs)
         except (SystemExit, error.TestBaseException):
             raise
         # If there was a different exception, turn it into a TestError.
         # It will be caught by step_engine or _run_step_fn.
         except Exception, e:
             raise error.UnhandledTestError(e)
 
 
     def cpu_count(self):
         return utils.count_cpus()  # use total system count
@@ skipping 611 matching lines @@
     assert myjob._record_indent == 0
 
     myjob.complete(0)
 
 
 site_job = utils.import_site_class(
     __file__, "autotest_lib.client.bin.site_job", "site_job", base_client_job)
 
 class job(site_job):
     pass