regression.py

Boost provides free peer-reviewed portable C++ source libraries. We emphasize libraries that work

        pass
    
    def command_update_source(self):
        if self.user and self.user != '' \
            or os.path.exists( os.path.join( self.boost_root, '.svn' ) ):
            open( self.timestamp_path, 'w' ).close()
            self.log( 'Updating sources from SVN (%s)...' % self.timestamp() )
            self.retry( self.svn_update )
        else:
            self.command_get_source( )
        pass
    
    def command_patch(self):
        self.import_utils()
        patch_boost_path = os.path.join( self.regression_root, self.patch_boost )
        if os.path.exists( patch_boost_path ):
            self.log( 'Found patch file "%s". Executing it.' % patch_boost_path )
            os.chdir( self.regression_root )
            utils.system( [ patch_boost_path ] )
        pass
    
    def command_setup(self):
        self.command_patch()
        self.build_if_needed(self.bjam,self.bjam_toolset)
        self.build_if_needed(self.process_jam_log,self.pjl_toolset)
    
    def command_test(self, *args):
        if not args or args == None or args == []: args = [ "test", "process" ]
        self.import_utils()

        self.log( 'Making "%s" directory...' % self.regression_results )
        utils.makedirs( self.regression_results )

        results_libs = os.path.join( self.regression_results, 'libs' )
        results_status = os.path.join( self.regression_results, 'status' )

        if "clean" in args:
            self.command_test_clean()

        if "test" in args:
            self.command_test_run()

        if "process" in args:
            self.command_test_process()
    
    def command_test_clean(self):
        results_libs = os.path.join( self.regression_results, 'libs' )
        results_status = os.path.join( self.regression_results, 'status' )
        self.rmtree( results_libs )
        self.rmtree( results_status )
    
    def command_test_run(self):
        self.import_utils()
        test_cmd = '%s -d2 --dump-tests %s "--build-dir=%s" >>"%s" 2>&1' % (
            self.bjam_cmd( self.toolsets ),
            self.bjam_options,
            self.regression_results,
            self.regression_log )
        self.log( 'Starting tests (%s)...' % test_cmd )
        cd = os.getcwd()
        os.chdir( os.path.join( self.boost_root, 'status' ) )
        utils.system( [ test_cmd ] )
        os.chdir( cd )

    def command_test_process(self):
        self.import_utils()
        self.log( 'Getting test case results out of "%s"...' % self.regression_log )
        cd = os.getcwd()
        os.chdir( os.path.join( self.boost_root, 'status' ) )
        utils.checked_system( [
            '"%s" "%s" <"%s"' % (
                self.tool_path(self.process_jam_log),
                self.regression_results,
                self.regression_log )
            ] )
        os.chdir( cd )
    
    def command_collect_logs(self):
        self.import_utils()
        comment_path = os.path.join( self.regression_root, self.comment )
        if not os.path.exists( comment_path ):
            self.log( 'Comment file "%s" not found; creating default comment.' % comment_path )
            f = open( comment_path, 'w' )
            f.write( '<p>Tests are run on %s platform.</p>' % self.platform_name() )
            f.close()

        if self.incremental:
            run_type = 'incremental'
        else:
            run_type = 'full'

        source = 'tarball'
        revision = ''
        svn_root_file = os.path.join( self.boost_root, '.svn' )
        svn_info_file = os.path.join( self.boost_root, 'svn_info.txt' )
        if os.path.exists( svn_root_file ):
            source = 'SVN'
            self.svn_command( 'info --xml "%s" >%s' % (self.boost_root,svn_info_file) )

        if os.path.exists( svn_info_file ):
            f = open( svn_info_file, 'r' )
            svn_info = f.read()
            f.close()
            i = svn_info.find( 'Revision:' )
            if i < 0: i = svn_info.find( 'revision=' )  # --xml format
            if i >= 0:
                i += 10
                while svn_info[i] >= '0' and svn_info[i] <= '9':
                  revision += svn_info[i]
                  i += 1

        from collect_and_upload_logs import collect_logs
        collect_logs(
            self.regression_results,
            self.runner, self.tag, self.platform, comment_path,
            self.timestamp_path,
            self.user,
            source, run_type,
            self.dart_server, self.proxy,
            revision )
        
    def command_upload_logs(self):
        self.import_utils()
        from collect_and_upload_logs import upload_logs
        self.retry(
            lambda:
                upload_logs(
                    self.regression_results,
                    self.runner, self.tag,
                    self.user,
                    self.ftp_proxy,
                    self.debug_level, self.send_bjam_log,
                    self.timestamp_path,
                    self.dart_server )
            )
    
    def command_regression(self):
        import socket
        import string
        try:
            mail_subject = 'Boost regression for %s on %s' % ( self.tag,
                string.split(socket.gethostname(), '.')[0] )
            start_time = time.localtime()
            if self.mail:
                self.log( 'Sending start notification to "%s"' % self.mail )
                self.send_mail(
                    '%s started at %s.' % ( mail_subject, format_time( start_time ) )
                    )
            
            self.command_get_tools()

            if self.local is not None:
                self.log( 'Using local file "%s"' % self.local )
                b = os.path.basename( self.local )
                tag = b[ 0: b.find( '.' ) ]
                self.log( 'Tag: "%s"' % tag  )
                self.unpack_tarball( local, self.boost_root )
                
            elif self.have_source:
                if not self.incremental: self.command_cleanup( [ 'bin' ] )
                
            else:
                if self.incremental or self.force_update:
                    if not self.incremental: self.command_cleanup( [ 'bin' ] )
                else:
                    self.command_cleanup()
                self.command_get_source()

            self.command_setup()

            # Not specifying --toolset in command line is not enough
            # that would mean to use Boost.Build default ones
            # We can skip test only we were explictly 
            # told to have no toolsets in command line "--toolset="
            if self.toolsets != '': # --toolset=,
                if not self.skip_tests:
                    self.command_test()
                self.command_collect_logs()
                self.command_upload_logs()

            if self.mail:
                self.log( 'Sending report to "%s"' % self.mail )
                end_time = time.localtime()
                self.send_mail(
                    '%s completed successfully at %s.' % ( mail_subject, format_time( end_time ) )
                    )
        except:
            if self.mail:
                self.log( 'Sending report to "%s"' % self.mail )
                traceback_ = '\n'.join( apply( traceback.format_exception, sys.exc_info() ) )
                end_time = time.localtime()
                self.send_mail(
                    '%s failed at %s.' % ( mail_subject, format_time( end_time ) ),
                    traceback_ )
            raise

    def command_show_revision(self):
        modified = '$Date: 2007-11-25 14:36:19 -0500 (Sun, 25 Nov 2007) $'
        revision = '$Revision: 41373 $'

        import re
        re_keyword_value = re.compile( r'^\$\w+:\s+(.*)\s+\$$' )
        print '\n\tRevision: %s' % re_keyword_value.match( revision ).group( 1 )
        print '\tLast modified on: %s\n' % re_keyword_value.match( modified ).group( 1 )
    
    #~ Utilities...
    
    def main(self):
        for action in self.actions:
            action_m = "command_"+action.replace('-','_')
            if hasattr(self,action_m):
                getattr(self,action_m)()

    def platform_name(self):
        # See http://article.gmane.org/gmane.comp.lib.boost.testing/933
        if sys.platform == 'win32':
            return 'Windows'
        elif sys.platform == 'cygwin':
            return 'Windows/Cygwin'
        return platform.system()

    def log(self,message):
        sys.stdout.flush()
        sys.stderr.flush()
        sys.stderr.write( '# %s\n' % message )
        sys.stderr.flush()

    def rmtree(self,path):
        if os.path.exists( path ):
            import shutil
            #~ shutil.rmtree( unicode( path ) )
            if sys.platform == 'win32':
                os.system( 'del /f /s /q "%s" >nul 2>&1' % path )
                shutil.rmtree( unicode( path ) )
            else:
                os.system( 'rm -f -r "%s"' % path )

    def refresh_timestamp( self ):
        if os.path.exists( self.timestamp_path ):
           os.unlink( self.timestamp_path )
        open( self.timestamp_path, 'w' ).close()

    def timestamp( self ):
        return time.strftime(
            '%Y-%m-%dT%H:%M:%SZ',
            time.gmtime( os.stat( self.timestamp_path ).st_mtime ) )

    def retry( self, f, max_attempts=5, sleep_secs=10 ):
        for attempts in range( max_attempts, -1, -1 ):
            try:
                return f()
            except Exception, msg:
                self.log( '%s failed with message "%s"' % ( f.__name__, msg ) )
                if attempts == 0:
                    self.log( 'Giving up.' )
                    raise

                self.log( 'Retrying (%d more attempts).' % attempts )
                time.sleep( sleep_secs )

    def http_get( self, source_url, destination_file ):
        import urllib
        
        proxies = None