Mercurial > gub
view bin/gub-tester @ 6481:c9f348420ef1
w32api: rename DATADIR to MINGW_DATADIR.
author | Jan Nieuwenhuizen <janneke@gnu.org> |
---|---|
date | Thu, 03 Feb 2011 19:55:51 +0100 |
parents | 72b46c22bf99 |
children |
line wrap: on
line source
#! /usr/bin/env python def argv0_relocation (): import os, sys bindir = os.path.dirname (sys.argv[0]) prefix = os.path.dirname (bindir) if not prefix: prefix = bindir + '/..' sys.path.insert (0, prefix) argv0_relocation () import sys import re import os import smtplib import email.MIMEText import email.Message import email.MIMEMultipart import optparse import time from gub.syntax import printf from gub import misc from gub import repository from gub import gub_log from gub.db import bsd as db_bsd def canonicalize_string (target): canonicalize = re.sub ('[ \t\n]', '_', target) canonicalize = re.sub ('[^a-zA-Z0-9-]+', '_', canonicalize) return canonicalize CACHED = 'CACHED' CACHED_FAIL = 'CACHED-FAIL' CACHED_SUCCESS = 'CACHED-SUCCESS' FAIL = 'FAIL' SUCCESS = 'SUCCESS' def result_message (parts, subject='') : """Concatenate PARTS to a Message object.""" if not parts: parts.append ('(empty)') parts = [email.MIMEText.MIMEText (p) for p in parts if p] msg = parts[0] if len (parts) > 1: msg = email.MIMEMultipart.MIMEMultipart () for p in parts: msg.attach (p) msg['Subject'] = subject msg.epilogue = '' return msg def opt_parser (): if 'EMAIL' in os.environ: address = os.environ['EMAIL'] else: try: address = '%s@localhost' % os.getlogin () except OSError: address = 'root@localhost' p = optparse.OptionParser (usage='gub-tester [options] command command ... ') p.address = [address] examples = """Examples: gub-tester --repository=downloads/lilypond.git \\ gub --branch=lilypond:master lilypond gub-tester --url=http://bazaar.launchpad.net/~yaffut/yaffut/yaffut.bzr \\ 'make all check' bzr branch http://bazaar.launchpad.net/~yaffut/yaffut/yaffut.bzr cd yaffut.bzr && gub-tester 'make all check' In a previously git-tester'ed directory: gub-tester --update 'make all check' """ misc.optparse_epilog (p, examples) p.add_option ('-t', '--to', action='append', dest='address', default=[], help='where to send error report') p.add_option ('--dry-run', dest='dry_run', default=False, action='store_true', help='do not run any commands') p.add_option ('--append-diff', dest='append_diff', default=False, action='store_true', help='append diff since last successful run') p.add_option ('--bcc', action='append', dest='bcc_address', default=[], help='BCC for error report') p.add_option ('-f', '--from', action='store', dest='sender', default=address, help='whom to list as sender') p.add_option ('--branch', action='store', dest='branch', default=None, help='which branch to fetch [GIT repositories]') p.add_option ('--url', action='store', dest='url', default=None, help='where to fetch sources') p.add_option ('--update', action='store_true', dest='update', default=False, help='checkout or update sources') p.add_option ('--revision', action='store', dest='revision', default=None, help='what revision to fetch') p.add_option ('--repository', action='store', dest='repository', default='.', help='where to download/cache repository') p.add_option ('--tag-repo', action='store', dest='tag_repo', default='', help='where to push success tags.') p.add_option ('--quiet', action='store_true', dest='be_quiet', default=False, help='only send mail when there was an error') p.add_option ('--force', action='store_true', dest='force', default=False, help='force rechecking even if release already checked') p.add_option ('--dependent', action='store_true', dest='is_dependent', default=False, help='test targets depend on each other') p.add_option ('--package', action='store', dest='package', default='lilypond', help='name of package under test') p.add_option ('--posthook', action='append', dest='posthooks', default=[], help='commands to execute after successful tests') p.add_option ('--test-self', action='store_true', dest='test_self', default=False, help='run a cursory self test') p.add_option ('-s', '--smtp', action='store', dest='smtp', default='localhost', help='SMTP server to use') p.add_option ('--result-directory', action='store', dest='result_dir', help='Where to store databases test results', default='log') p.add_option ('-v', '--verbose', action='count', dest='verbosity', default=0) return p def get_db (options, name): name = options.result_dir + '/%s.db' % name db_file = os.path.join (options.result_dir, name) db = db_bsd.open (db_file, 'c') return db def test_target (repo, options, target, last_patch): canonicalize = canonicalize_string (target) release_hash = repo.get_checksum () done_db = get_db (options, canonicalize) tag_db = repository.TagDb (options.result_dir) recheck = '' if release_hash in done_db: gub_log.info ('release %(release_hash)s has already been checked\n' % locals ()) if not options.force: RESULT = CACHED_SUCCESS if done_db[release_hash].endswith (FAIL): RESULT = CACHED_FAIL return RESULT, ['cached %(RESULT)s of %(release_hash)s' % locals ()] recheck = ' (forced recheck of %(release_hash)s)' % locals () gub_log.info ('recheck forced\n' % locals ()) log = 'test-%(canonicalize)s.log' % locals () log = os.path.join (options.result_dir, log) cmd = 'nice time sh -c "(%(target)s)"' % locals () gub_log.command (cmd + '\n') gub_log.error ('Logging test output to %(log)s\n' % locals ()) if options.dry_run: return (SUCCESS, ['dryrun']) gub_log.set_default_log (log, options.verbosity) logger = gub_log.default_logger stat = loggedos.system (logger, cmd, ignore_errors=True) base_tag = 'success-%(canonicalize)s-' % locals () result = 'unknown' attachments = [] body = test_log.read_tail () diff = None if stat: RESULT = FAIL result = 'error' if options.append_diff: if repo.is_distributed (): diff = repo.get_diff_from_tag_base (base_tag) else: diff = tag_db.get_diff_from_tag_base (base_tag, repo) else: RESULT = SUCCESS result = 'success' body = body[-10:] tag_db.tag (base_tag, repo) if repo.is_distributed (): tag = repo.tag (base_tag) gub_log.action ('tagging with %(tag)s\n' % locals ()) if options.tag_repo: repo.push (tag, options.tag_repo) message = '''%(result)s for%(recheck)s: %(target)s %(tail)s''' tail = '\n'.join (body) attachments = [message % locals ()] if diff: attachments.append (diff) gub_log.info ('%(target)s: %(RESULT)s\n' % locals ()) done_db[release_hash] = time.ctime () + ' ' + RESULT return (RESULT, attachments) def send_message (options, msg): if not options.address: gub_log.info ('No recipients for result mail\n') # FIXME: what about env[EMAIL]? return False COMMASPACE = ', ' msg['From'] = options.sender msg['To'] = COMMASPACE.join (options.address) if options.bcc_address: msg['BCC'] = COMMASPACE.join (options.bcc_address) msg['X-Autogenerated'] = 'GUB %s' % options.package connection = smtplib.SMTP (options.smtp) connection.sendmail (options.sender, options.address, msg.as_string ()) return True def send_result_by_mail (options, parts, subject='Autotester result'): msg = result_message (parts, subject) if not send_message (options, msg): print_results (options, parts, subject) def print_results (options, parts, subject='Autotester result'): printf ('\n===\n\nSUBJECT: ', subject) printf ('\n---\n'.join (parts)) printf ('END RESULT\n===') def real_main (options, args, handle_result): log = os.path.join (options.result_dir, 'gub-tester.log') if options.dry_run: log = '/dev/stdout' gub_log.set_default_log (log, options.verbosity) gub_log.info (' *** %s\n' % time.ctime ()) gub_log.info (' *** Starting tests:\n %s\n' % '\n '.join (args)) repo = repository.get_repository_proxy (options.repository, options.url, options.branch, options.revision) if ((not repo.is_downloaded () or options.update or options.revision) and repo.source): repo.download () repo.update_workdir ('.') gub_log.info ('Repository %s\n' % str (repo)) last_patch = repo.get_revision_description () release_hash = repo.get_checksum () release_id = ''' Last patch of this release: %(last_patch)s\n Checksum of revision: %(release_hash)s ''' % locals () summary_body = '\n\n' results = {} failures = 0 cached_failures = 0 for a in args: result_tup = test_target (repo, options, a, last_patch) (result, atts) = result_tup if result.startswith (CACHED): if result == CACHED_FAIL: cached_failures += 1 continue results[a] = result_tup success = result.startswith (SUCCESS) if not (options.be_quiet and success): handle_result (options, atts, subject='Autotester: %s %s' % (result, a)) summary_body += '%s\n %s\n' % (a, result) if not success: failures += 1 if options.is_dependent: break if (results and len (args) > 1 and (failures > 0 or not options.be_quiet)): handle_result (options, [summary_body, release_id], subject='Autotester: summary') if failures == 0 and results: for p in options.posthooks: os.system (p) return failures + cached_failures def test_self (options, args): def system (c): printf (c) if os.system (c): raise Exception ('barf') self_test_dir = 'test-gub-test.darcs' system ('rm -rf %s ' % self_test_dir) system ('mkdir %s ' % self_test_dir) os.chdir (self_test_dir) system ('mkdir log') system ('''echo '#!/bin/sh true' > foo.sh''') system ('darcs init') system ('echo author > _darcs/prefs/author') system ('darcs add foo.sh') system ('darcs record -am "add bla"') options.repository = os.getcwd () real_main (options, ['false', 'true', 'sh foo.sh'], print_results) system ('''echo '#!/bin/sh true' > foo.sh''') system ('darcs record -am "change bla"') real_main (options, ['sh foo.sh'], print_results) def main (): p = opt_parser () (options, args) = p.parse_args () if not args: gub_log.error ('error: nothing to do\n') p.print_help () sys.exit (2) if not os.path.isdir (options.result_dir): os.makedirs (options.result_dir) options.result_dir = os.path.abspath (options.result_dir) status = 0 if options.test_self: status = test_self (options, args) else: status = real_main (options, args, send_result_by_mail) sys.exit (status) if __name__ == '__main__': main ()