# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import httplib
import json
import os
import platform
import random
import re
import socket
import tempfile
import time
import traceback
import urllib
from threading import RLock
from mozprofile import Profile
from tps.firefoxrunner import TPSFirefoxRunner
from tps.phase import TPSTestPhase
from tps.mozhttpd import MozHttpd
class TempFile(object):
"""Class for temporary files that delete themselves when garbage-collected.
"""
def __init__(self, prefix=None):
self.fd, self.filename = self.tmpfile = tempfile.mkstemp(prefix=prefix)
def write(self, data):
if self.fd:
os.write(self.fd, data)
def close(self):
if self.fd:
os.close(self.fd)
self.fd = None
def cleanup(self):
if self.fd:
self.close()
if os.access(self.filename, os.F_OK):
os.remove(self.filename)
__del__ = cleanup
class TPSTestRunner(object):
default_env = { 'MOZ_CRASHREPORTER_DISABLE': '1',
'GNOME_DISABLE_CRASH_DIALOG': '1',
'XRE_NO_WINDOWS_CRASH_DIALOG': '1',
'MOZ_NO_REMOTE': '1',
'XPCOM_DEBUG_BREAK': 'warn',
}
default_preferences = { 'app.update.enabled' : False,
'extensions.getAddons.get.url': 'http://127.0.0.1:4567/en-US/firefox/api/%API_VERSION%/search/guid:%IDS%',
'extensions.update.enabled' : False,
'extensions.update.notifyUser' : False,
'browser.shell.checkDefaultBrowser' : False,
'browser.tabs.warnOnClose' : False,
'browser.warnOnQuit': False,
'browser.sessionstore.resume_from_crash': False,
'services.sync.addons.ignoreRepositoryChecking': True,
'services.sync.firstSync': 'notReady',
'services.sync.lastversion': '1.0',
'services.sync.log.rootLogger': 'Trace',
'services.sync.log.logger.engine.addons': 'Trace',
'services.sync.log.logger.service.main': 'Trace',
'services.sync.log.logger.engine.bookmarks': 'Trace',
'services.sync.log.appender.console': 'Trace',
'services.sync.log.appender.debugLog.enabled': True,
'toolkit.startup.max_resumed_crashes': -1,
'browser.dom.window.dump.enabled': True,
# Allow installing extensions dropped into the profile folder
'extensions.autoDisableScopes': 10,
# Don't open a dialog to show available add-on updates
'extensions.update.notifyUser' : False,
}
syncVerRe = re.compile(
r"Sync version: (?P<syncversion>.*)\n")
ffVerRe = re.compile(
r"Firefox version: (?P<ffver>.*)\n")
ffDateRe = re.compile(
r"Firefox builddate: (?P<ffdate>.*)\n")
def __init__(self, extensionDir, emailresults=False, testfile="sync.test",
binary=None, config=None, rlock=None, mobile=False,
autolog=False, logfile="tps.log",
ignore_unused_engines=False):
self.extensions = []
self.emailresults = emailresults
self.testfile = testfile
self.logfile = os.path.abspath(logfile)
self.binary = binary
self.ignore_unused_engines = ignore_unused_engines
self.config = config if config else {}
self.repo = None
self.changeset = None
self.branch = None
self.numfailed = 0
self.numpassed = 0
self.nightly = False
self.rlock = rlock
self.mobile = mobile
self.autolog = autolog
self.tpsxpi = None
self.firefoxRunner = None
self.extensionDir = extensionDir
self.productversion = None
self.addonversion = None
self.postdata = {}
self.errorlogs = {}
@property
def mobile(self):
return self._mobile
@mobile.setter
def mobile(self, value):
self._mobile = value
self.synctype = 'desktop' if not self._mobile else 'mobile'
def log(self, msg, printToConsole=False):
"""Appends a string to the logfile"""
f = open(self.logfile, 'a')
f.write(msg)
f.close()
if printToConsole:
print msg
def _zip_add_file(self, zip, file, rootDir):
zip.write(os.path.join(rootDir, file), file)
def _zip_add_dir(self, zip, dir, rootDir):
try:
zip.write(os.path.join(rootDir, dir), dir)
except:
# on some OS's, adding directory entries doesn't seem to work
pass
for root, dirs, files in os.walk(os.path.join(rootDir, dir)):
for f in files:
zip.write(os.path.join(root, f), os.path.join(dir, f))
def run_single_test(self, testdir, testname):
testpath = os.path.join(testdir, testname)
self.log("Running test %s\n" % testname)
# Create a random account suffix that is used when creating test
# accounts on a staging server.
account_suffix = {"account-suffix": ''.join([str(random.randint(0,9))
for i in range(1,6)])}
self.config['account'].update(account_suffix)
# Read and parse the test file, merge it with the contents of the config
# file, and write the combined output to a temporary file.
f = open(testpath, 'r')
testcontent = f.read()
f.close()
try:
test = json.loads(testcontent)
except:
test = json.loads(testcontent[testcontent.find("{"):testcontent.find("}") + 1])
testcontent += 'var config = %s;\n' % json.dumps(self.config, indent=2)
testcontent += 'var seconds_since_epoch = %d;\n' % int(time.time())
tmpfile = TempFile(prefix='tps_test_')
tmpfile.write(testcontent)
tmpfile.close()
# generate the profiles defined in the test, and a list of test phases
profiles = {}
phaselist = []
for phase in test:
profilename = test[phase]
# create the profile if necessary
if not profilename in profiles:
profiles[profilename] = Profile(preferences = self.preferences,
addons = self.extensions)
# create the test phase
phaselist.append(TPSTestPhase(
phase,
profiles[profilename],
testname,
tmpfile.filename,
self.logfile,
self.env,
self.firefoxRunner,
self.log,
ignore_unused_engines=self.ignore_unused_engines))
# sort the phase list by name
phaselist = sorted(phaselist, key=lambda phase: phase.phase)
# run each phase in sequence, aborting at the first failure
for phase in phaselist:
phase.run()
# if a failure occurred, dump the entire sync log into the test log
if phase.status != "PASS":
for profile in profiles:
self.log("\nDumping sync log for profile %s\n" % profiles[profile].profile)
for root, dirs, files in os.walk(os.path.join(profiles[profile].profile, 'weave', 'logs')):
for f in files:
weavelog = os.path.join(profiles[profile].profile, 'weave', 'logs', f)
if os.access(weavelog, os.F_OK):
with open(weavelog, 'r') as fh:
for line in fh:
possible_time = line[0:13]
if len(possible_time) == 13 and possible_time.isdigit():
time_ms = int(possible_time)
formatted = time.strftime('%Y-%m-%d %H:%M:%S',
time.localtime(time_ms / 1000))
self.log('%s.%03d %s' % (
formatted, time_ms % 1000, line[14:] ))
else:
self.log(line)
break;
# grep the log for FF and sync versions
f = open(self.logfile)
logdata = f.read()
match = self.syncVerRe.search(logdata)
sync_version = match.group("syncversion") if match else 'unknown'
match = self.ffVerRe.search(logdata)
firefox_version = match.group("ffver") if match else 'unknown'
match = self.ffDateRe.search(logdata)
firefox_builddate = match.group("ffdate") if match else 'unknown'
f.close()
if phase.status == 'PASS':
logdata = ''
else:
# we only care about the log data for this specific test
logdata = logdata[logdata.find('Running test %s' % (str(testname))):]
result = {
'PASS': lambda x: ('TEST-PASS', ''),
'FAIL': lambda x: ('TEST-UNEXPECTED-FAIL', x.rstrip()),
'unknown': lambda x: ('TEST-UNEXPECTED-FAIL', 'test did not complete')
} [phase.status](phase.errline)
logstr = "\n%s | %s%s\n" % (result[0], testname, (' | %s' % result[1] if result[1] else ''))
try:
repoinfo = self.firefoxRunner.runner.get_repositoryInfo()
except:
repoinfo = {}
apprepo = repoinfo.get('application_repository', '')
appchangeset = repoinfo.get('application_changeset', '')
# save logdata to a temporary file for posting to the db
tmplogfile = None
if logdata:
tmplogfile = TempFile(prefix='tps_log_')
tmplogfile.write(logdata)
tmplogfile.close()
self.errorlogs[testname] = tmplogfile
resultdata = ({ "productversion": { "version": firefox_version,
"buildid": firefox_builddate,
"builddate": firefox_builddate[0:8],
"product": "Firefox",
"repository": apprepo,
"changeset": appchangeset,
},
"addonversion": { "version": sync_version,
"product": "Firefox Sync" },
"name": testname,
"message": result[1],
"state": result[0],
"logdata": logdata
})
self.log(logstr, True)
for phase in phaselist:
print "\t%s: %s" % (phase.phase, phase.status)
if phase.status == 'FAIL':
break
return resultdata
def run_tests(self):
# delete the logfile if it already exists
if os.access(self.logfile, os.F_OK):
os.remove(self.logfile)
# Make a copy of the default env variables and preferences, and update
# them for mobile settings if needed.
self.env = self.default_env.copy()
self.preferences = self.default_preferences.copy()
if self.mobile:
self.preferences.update({'services.sync.client.type' : 'mobile'})
# Acquire a lock to make sure no other threads are running tests
# at the same time.
if self.rlock:
self.rlock.acquire()
try:
# Create the Firefox runner, which will download and install the
# build, as needed.
if not self.firefoxRunner:
self.firefoxRunner = TPSFirefoxRunner(self.binary)
# now, run the test group
self.run_test_group()
except:
traceback.print_exc()
self.numpassed = 0
self.numfailed = 1
if self.emailresults:
try:
self.sendEmail('<pre>%s</pre>' % traceback.format_exc(),
sendTo='crossweave@mozilla.com')
except:
traceback.print_exc()
else:
raise
else:
try:
if self.autolog:
self.postToAutolog()
if self.emailresults:
self.sendEmail()
except:
traceback.print_exc()
try:
self.sendEmail('<pre>%s</pre>' % traceback.format_exc(),
sendTo='crossweave@mozilla.com')
except:
traceback.print_exc()
# release our lock
if self.rlock:
self.rlock.release()
# dump out a summary of test results
print 'Test Summary\n'
for test in self.postdata.get('tests', {}):
print '%s | %s | %s' % (test['state'], test['name'], test['message'])
def run_test_group(self):
self.results = []
self.extensions = []
# set the OS we're running on
os_string = platform.uname()[2] + " " + platform.uname()[3]
if os_string.find("Darwin") > -1:
os_string = "Mac OS X " + platform.mac_ver()[0]
if platform.uname()[0].find("Linux") > -1:
os_string = "Linux " + platform.uname()[5]
if platform.uname()[0].find("Win") > -1:
os_string = "Windows " + platform.uname()[3]
# reset number of passed/failed tests
self.numpassed = 0
self.numfailed = 0
# build our tps.xpi extension
self.extensions.append(os.path.join(self.extensionDir, 'tps'))
self.extensions.append(os.path.join(self.extensionDir, "mozmill"))
# build the test list
try:
f = open(self.testfile)
jsondata = f.read()
f.close()
testfiles = json.loads(jsondata)
testlist = testfiles['tests']
except ValueError:
testlist = [os.path.basename(self.testfile)]
testdir = os.path.dirname(self.testfile)
self.mozhttpd = MozHttpd(port=4567, docroot=testdir)
self.mozhttpd.start()
# run each test, and save the results
for test in testlist:
result = self.run_single_test(testdir, test)
if not self.productversion:
self.productversion = result['productversion']
if not self.addonversion:
self.addonversion = result['addonversion']
self.results.append({'state': result['state'],
'name': result['name'],
'message': result['message'],
'logdata': result['logdata']})
if result['state'] == 'TEST-PASS':
self.numpassed += 1
else:
self.numfailed += 1
self.mozhttpd.stop()
# generate the postdata we'll use to post the results to the db
self.postdata = { 'tests': self.results,
'os':os_string,
'testtype': 'crossweave',
'productversion': self.productversion,
'addonversion': self.addonversion,
'synctype': self.synctype,
}
def sendEmail(self, body=None, sendTo=None):
# send the result e-mail
if self.config.get('email') and self.config['email'].get('username') \
and self.config['email'].get('password'):
from tps.sendemail import SendEmail
from tps.emailtemplate import GenerateEmailBody
if body is None:
buildUrl = None
if self.firefoxRunner and self.firefoxRunner.url:
buildUrl = self.firefoxRunner.url
body = GenerateEmailBody(self.postdata,
self.numpassed,
self.numfailed,
self.config['account']['serverURL'],
buildUrl)
subj = "TPS Report: "
if self.numfailed == 0 and self.numpassed > 0:
subj += "YEEEAAAHHH"
else:
subj += "PC LOAD LETTER"
changeset = self.postdata['productversion']['changeset'] if \
self.postdata and self.postdata.get('productversion') and \
self.postdata['productversion'].get('changeset') \
else 'unknown'
subj +=", changeset " + changeset + "; " + str(self.numfailed) + \
" failed, " + str(self.numpassed) + " passed"
To = [sendTo] if sendTo else None
if not To:
if self.numfailed > 0 or self.numpassed == 0:
To = self.config['email'].get('notificationlist')
else:
To = self.config['email'].get('passednotificationlist')
if To:
SendEmail(From=self.config['email']['username'],
To=To,
Subject=subj,
HtmlData=body,
Username=self.config['email']['username'],
Password=self.config['email']['password'])
def postToAutolog(self):
from mozautolog import RESTfulAutologTestGroup as AutologTestGroup
group = AutologTestGroup(
harness='crossweave',
testgroup='crossweave-%s' % self.synctype,
server=self.config.get('es'),
restserver=self.config.get('restserver'),
machine=socket.gethostname(),
platform=self.config.get('platform', None),
os=self.config.get('os', None),
)
tree = self.postdata['productversion']['repository']
group.set_primary_product(
tree=tree[tree.rfind("/")+1:],
version=self.postdata['productversion']['version'],
buildid=self.postdata['productversion']['buildid'],
buildtype='opt',
revision=self.postdata['productversion']['changeset'],
)
group.add_test_suite(
passed=self.numpassed,
failed=self.numfailed,
todo=0,
)
for test in self.results:
if test['state'] != "TEST-PASS":
errorlog = self.errorlogs.get(test['name'])
errorlog_filename = errorlog.filename if errorlog else None
group.add_test_failure(
test = test['name'],
status = test['state'],
text = test['message'],
logfile = errorlog_filename
)
try:
group.submit()
except:
self.sendEmail('<pre>%s</pre>' % traceback.format_exc(),
sendTo='crossweave@mozilla.com')
return
# Iterate through all testfailure objects, and update the postdata
# dict with the testfailure logurl's, if any.
for tf in group.testsuites[-1].testfailures:
result = [x for x in self.results if x.get('name') == tf.test]
if not result:
continue
result[0]['logurl'] = tf.logurl
