Commit 4514c00d authored by Jérome Perrin's avatar Jérome Perrin

recipe: drop slapos.recipe:seleniumrunner

We use jstestnode
parent 26ecb4d2
......@@ -166,7 +166,6 @@ setup(name=name,
'slapos.recipe.request:RequestOptionalJSONEncoded',
're6stnet.registry = slapos.recipe.re6stnet:Recipe',
'reverseproxy.nginx = slapos.recipe.reverse_proxy_nginx:Recipe',
'seleniumrunner = slapos.recipe.seleniumrunner:Recipe',
'sheepdogtestbed = slapos.recipe.sheepdogtestbed:SheepDogTestBed',
'shell = slapos.recipe.shell:Recipe',
'shellinabox = slapos.recipe.shellinabox:Recipe',
......
seleniumrunner
==============
Allows to run selenium tests through browser and xvfb. Posts the results on
Nexedi ERP5.
Parameters
==========
* project : name of the project inside of ERP5 test result instance
* user : username to use in ERP5 instance to test
* password : password to use in ERP5 instance to test
* suite_name : name of test suite to launch
* url : url to portal_test of ERP5 isntance to test
* test_report_instance_url : url of test_result_module to put results
* Example::
<?xml version="1.0" encoding="utf-8"?>
<instance>
<parameter id="project">Vifib</parameter>
<parameter id="user">myuser</parameter>
<parameter id="password">mypassword</parameter>
<parameter id="suite_name">my_zuite</parameter>
<parameter id="url">https://myerp5totest/erp5/portal_tests</parameter>
<parameter id="test_report_instance_url">https://user:passwordwww.myerp5withtestresults.com/test_result_module/</parameter>
</instance>
##############################################################################
#
# Copyright (c) 2011 Nexedi SARL and Contributors. All Rights Reserved.
# Kazuhiko <kazuhiko@nexedi.com>
# Rafael Monnerat <rafael@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# garantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
# XXX-Cedric : This is an import of
# https://lab.nexedi.com/nexedi/erp5/blob/HEAD/product/ERP5Type/tests/ERP5TypeFunctionalTestCase.py
# Modification of the present file should be ported back to this original file.
import os
import time
import signal
import re
from subprocess import Popen, PIPE
import shutil
#from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase, \
# _getConversionServerDict
# REGEX FOR ZELENIUM TESTS
TEST_PASS_RE = re.compile('<th[^>]*>Tests passed</th>\n\s*<td[^>]*>([^<]*)')
TEST_FAILURE_RE = re.compile('<th[^>]*>Tests failed</th>\n\s*<td[^>]*>([^<]*)')
IMAGE_RE = re.compile('<img[^>]*?>')
TEST_ERROR_TITLE_RE = re.compile('(?:error.gif.*?>|title status_failed"><td[^>]*>)([^>]*?)</td></tr>', re.S)
TEST_RESULT_RE = re.compile('<div style="padding-top: 10px;">\s*<p>\s*'
'<img.*?</div>\s.*?</div>\s*', re.S)
TEST_ERROR_RESULT_RE = re.compile('.*(?:error.gif|title status_failed).*', re.S)
ZELENIUM_BASE_URL = "%s/portal_tests/%s/core/TestRunner.html?test=../test_suite_html&auto=on&resultsUrl=%s/portal_tests/postResults&__ac_name=%s&__ac_password=%s"
tests_framework_home = os.path.dirname(os.path.abspath(__file__))
# handle 'system global' instance
if tests_framework_home.startswith('/usr/lib'):
real_instance_home = '/var/lib/erp5'
else:
real_instance_home = os.path.sep.join(
tests_framework_home.split(os.path.sep)[:-3])
instance_home = os.path.join(real_instance_home, 'unit_test')
bt5_dir_list = ','.join([
os.path.join(instance_home, 'Products/ERP5/bootstrap'),
os.path.join(instance_home, 'bt5')])
class TimeoutError(Exception):
pass
class Xvfb:
def __init__(self, fbdir, xvfb_location="Xvfb"):
self.display_list = [":%s" % i for i in range(123, 144)]
self.display = None
self.fbdir = fbdir
self.pid = None
self.xvfb_location = xvfb_location
def _runCommand(self, display):
command = [self.xvfb_location, '-fbdir' , self.fbdir, display]
self.process = Popen(" ".join(command),
stdout=PIPE,
stderr=PIPE,
shell=True,
close_fds=True)
def run(self):
for display_try in self.display_list:
lock_filepath = '/tmp/.X%s-lock' % display_try.replace(":", "")
if not os.path.exists(lock_filepath):
self._runCommand(display_try)
self.display = display_try
break
#display = os.environ.get('DISPLAY')
#if display:
# auth = Popen(['xauth', 'list', display], stdout=PIPE).communicate()[0]
# if auth:
# (displayname, protocolname, hexkey) = auth.split()
# Popen(['xauth', 'add', 'localhost/unix:%s' % display, protocolname, hexkey])
print 'Xvfb : %d' % self.process.pid
print 'Take screenshots using xwud -in %s/Xvfb_screen0' % self.fbdir
def quit(self):
if hasattr(self, 'process'):
process_pid = self.process.pid
try:
self.process.terminate()
finally:
if process_pid:
print "Stopping Xvfb on pid: %s" % self.pid
os.kill(process_pid, signal.SIGTERM)
class Browser:
use_xvfb = 1
def __init__(self, profile_dir, host, bin_location=None):
self.bin_location = bin_location
self.profile_dir = profile_dir
self.host = host
self.pid = None
def quit(self):
if self.pid:
os.kill(self.pid, signal.SIGTERM)
def _run(self, url, display):
""" This method should be implemented on a subclass """
raise NotImplementedError
def _setEnviron(self):
pass
def run(self, url, display):
self.clean()
self._setEnviron()
self._setDisplay(display)
self._run(url)
print "Browser %s running on pid: %s" % (self.__class__.__name__, self.pid)
def clean(self):
""" Clean up removing profile dir and recreating it"""
os.system("rm -rf %s" % self.profile_dir)
os.mkdir(self.profile_dir)
def _createFile(self, filename, content):
file_path = os.path.join(self.profile_dir, filename)
with open(file_path, 'w') as f:
f.write(content)
return file_path
def _setDisplay(self, display):
if display is None:
try:
shutil.copy2(os.path.expanduser('~/.Xauthority'), '%s/.Xauthority' % self.profile_dir)
except IOError:
pass
else:
os.environ["DISPLAY"] = display
def _runCommand(self, command_tuple):
print " ".join(list(command_tuple))
self.pid = os.spawnlp(os.P_NOWAIT, *command_tuple)
class Firefox(Browser):
""" Use firefox to open run all the tests"""
def _setEnviron(self):
os.environ['MOZ_NO_REMOTE'] = '1'
os.environ['HOME'] = self.profile_dir
os.environ['LC_ALL'] = 'C'
os.environ["MOZ_CRASHREPORTER_DISABLE"] = "1"
os.environ["NO_EM_RESTART"] = "1"
def _run(self, url):
# Prepare to run
if not self.bin_location:
self.bin_location = "firefox"
self._createFile('prefs.js', self.getPrefJs())
self._runCommand((self.bin_location, "firefox", "-no-remote",
"-profile", self.profile_dir, url))
os.environ['MOZ_NO_REMOTE'] = '0'
def getPrefJs(self):
return """
// Don't ask if we want to switch default browsers
user_pref("browser.shell.checkDefaultBrowser", false);
// Disable pop-up blocking
user_pref("browser.allowpopups", true);
user_pref("dom.disable_open_during_load", false);
// Configure us as the local proxy
//user_pref("network.proxy.type", 2);
// Disable security warnings
user_pref("security.warn_submit_insecure", false);
user_pref("security.warn_submit_insecure.show_once", false);
user_pref("security.warn_entering_secure", false);
user_pref("security.warn_entering_secure.show_once", false);
user_pref("security.warn_entering_weak", false);
user_pref("security.warn_entering_weak.show_once", false);
user_pref("security.warn_leaving_secure", false);
user_pref("security.warn_leaving_secure.show_once", false);
user_pref("security.warn_viewing_mixed", false);
user_pref("security.warn_viewing_mixed.show_once", false);
// Disable "do you want to remember this password?"
user_pref("signon.rememberSignons", false);
// increase the timeout before warning of unresponsive script
user_pref("dom.max_script_run_time", 120);
// this is required to upload files
user_pref("capability.principal.codebase.p1.granted", "UniversalFileRead");
user_pref("signed.applets.codebase_principal_support", true);
user_pref("capability.principal.codebase.p1.id", "http://%s");
user_pref("capability.principal.codebase.p1.subjectName", "");""" % \
(self.host)
class PhantomJS(Browser):
def _createRunJS(self):
run_js = """
var page = new WebPage(),
address;
address = phantom.args[0];
page.open(address, function (status) {
if (status !== 'success') {
console.log('FAIL to load the address');
} else {
console.log('SUCCESS load the address');
}
phantom.exit();
});
"""
return self._createFile('run.js', run_js)
def _run(self, url):
self._runCommand(("phantomjs", "phantomjs", self._createRunJS(), url))
class FunctionalTestRunner:
# There is no test that can take more them 24 hours
timeout = 2.0 * 60 * 60
def __init__(self, host, port, portal, run_only='', use_phanthom=False):
self.instance_home = os.environ['INSTANCE_HOME']
# Such informations should be automatically loaded
self.user = 'ERP5TypeTestCase'
self.password = ''
self.run_only = run_only
profile_dir = os.path.join(self.instance_home, 'profile')
self.portal = portal
if use_phanthom:
self.browser = PhantomJS(profile_dir, host, int(port))
else:
self.browser = Firefox(profile_dir, host, int(port))
def getStatus(self):
transaction.commit()
return self.portal.portal_tests.TestTool_getResults()
def _getTestURL(self):
return ZELENIUM_BASE_URL % (self.portal.portal_url(), self.run_only,
self.portal.portal_url(), self.user, self.password)
def test(self, debug=0):
xvfb = Xvfb(self.instance_home)
try:
start = time.time()
if not debug and self.browser.use_xvfb:
xvfb.run()
self.browser.run(self._getTestURL() , xvfb.display)
while self.getStatus() is None:
time.sleep(10)
if (time.time() - start) > float(self.timeout):
raise TimeoutError("Test took more them %s seconds" % self.timeout)
finally:
self.browser.quit()
xvfb.quit()
def processResult(self):
file_content = self.getStatus().encode("utf-8", "replace")
sucess_amount = TEST_PASS_RE.search(file_content).group(1)
failure_amount = TEST_FAILURE_RE.search(file_content).group(1)
error_title_list = [re.compile('\s+').sub(' ', x).strip()
for x in TEST_ERROR_TITLE_RE.findall(file_content)]
detail = ''
for test_result in TEST_RESULT_RE.findall(file_content):
if TEST_ERROR_RESULT_RE.match(test_result):
detail += test_result
detail = IMAGE_RE.sub('', detail)
if detail:
detail = IMAGE_RE.sub('', detail)
detail = '''<html>
<head>
<style type="text/css">tr.status_failed { background-color:red };</style>
</head>
<body>%s</body>
</html>''' % detail
# When return fix output for handle unicode issues.
return detail, int(sucess_amount), int(failure_amount), error_title_list
class ERP5TypeFunctionalTestCase:
run_only = ""
foreground = 0
use_phanthom = False
def getTitle(self):
return "Zelenium"
def afterSetUp(self):
# create browser_id_manager
if not "browser_id_manager" in self.portal.objectIds():
self.portal.manage_addProduct['Sessions'].constructBrowserIdManager()
transaction.commit()
self.setSystemPreference()
self.portal.portal_tests.TestTool_cleanUpTestResults()
self.stepTic()
def setSystemPreference(self):
conversion_dict = _getConversionServerDict()
self.portal.Zuite_setPreference(
working_copy_list=bt5_dir_list,
conversion_server_hostname=conversion_dict['hostname'],
conversion_server_port=conversion_dict['port']
)
# XXX Memcached is missing
# XXX Persistent cache setup is missing
def testFunctionalTestRunner(self):
# first of all, abort to get rid of the mysql participation inn this
# transaction
self.portal._p_jar.sync()
self.runner = FunctionalTestRunner(self.serverhost, self.serverport,
self.portal, self.run_only, self.use_phanthom)
self.runner.test(debug=self.foreground)
detail, success, failure, error_title_list = self.runner.processResult()
self.logMessage("-" * 79)
total = success + failure
self.logMessage("%s Functional Tests %s Tests, %s Failures" % \
(self.getTitle(), total, failure))
self.logMessage("-" * 79)
self.logMessage(detail)
self.logMessage("-" * 79)
self.assertEquals([], error_title_list)
##############################################################################
#
# Copyright (c) 2010 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
#############################################################################
from slapos.recipe.librecipe import GenericBaseRecipe
class Recipe(GenericBaseRecipe):
def install(self):
return self.createPythonScript(
self.options['runner-path'],
__name__+'.testrunner.run',
(self.options['suite-url'],
self.options['report-url'],
self.options['report-project'],
self.options['browser']))
##############################################################################
#
# Copyright (c) 2010 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
#############################################################################
from __future__ import absolute_import
import re
import urlparse
import urllib
import httplib
import mimetools
from random import randint
import tempfile
import os
import stat
import zipfile
import mimetypes
import datetime
TB_SEP = "============================================================="\
"========="
# REGEX FOR ZELENIUM TESTS
TEST_PASS_RE = re.compile('<th[^>]*>Tests passed</th>\n\s*<td[^>]*>([^<]*)')
TEST_FAILURE_RE = re.compile('<th[^>]*>Tests failed</th>\n\s*<td[^>]*>([^<]*)')
IMAGE_RE = re.compile('<img[^>]*?>')
TEST_ERROR_TITLE_RE = re.compile('(?:error.gif.*?>|title status_failed"><td[^>]*>)([^>]*?)</td></tr>', re.S)
TEST_RESULT_RE = re.compile('<div style="padding-top: 10px;">\s*<p>\s*'
'<img.*?</div>\s.*?</div>\s*', re.S)
DURATION_RE = re.compile('<th[^>]*>Elapsed time \(sec\)</th>\n\s*<td[^>]*>([^<]*)')
TEST_ERROR_RESULT_RE = re.compile('.*(?:error.gif|title status_failed).*', re.S)
def get_content_type(f):
return mimetypes.guess_type(f.name)[0] or 'application/octet-stream'
class ConnectionHelper:
def __init__(self, url):
self.conn = urlparse.urlparse(url)
if self.conn.scheme == 'http':
connection_type = httplib.HTTPConnection
if self.conn.port is None:
self.port = 80
else:
connection_type = httplib.HTTPSConnection
if self.conn.port is None:
self.port = 443
self.connection_type = connection_type
def _connect(self):
self.connection = self.connection_type(self.conn.hostname + ':' +
str(self.conn.port or self.port))
def POST(self, path, parameter_dict, file_list=None):
self._connect()
parameter_dict.update(__ac_name=self.conn.username,
__ac_password=self.conn.password)
header_dict = {'Content-type': "application/x-www-form-urlencoded"}
if file_list is None:
body = urllib.urlencode(parameter_dict)
else:
boundary = mimetools.choose_boundary()
header_dict['Content-type'] = 'multipart/form-data; boundary=%s' % (
boundary,)
body = ''
for k, v in parameter_dict.iteritems():
body += '--%s\r\n' % boundary
body += 'Content-Disposition: form-data; name="%s"\r\n' % k
body += '\r\n'
body += '%s\r\n' % v
for name, filename in file_list:
f = open(filename, 'r')
body += '--%s\r\n' % boundary
body += 'Content-Disposition: form-data; name="%s"; filename="%s"\r\n'\
% (name, name)
body += 'Content-Type: %s\r\n' % get_content_type(f)
body += 'Content-Length: %d\r\n' % os.fstat(f.fileno())[stat.ST_SIZE]
body += '\r\n'
body += f.read()
f.close()
body += '\r\n'
self.connection.request("POST", self.conn.path + '/' + path,
body, header_dict)
self.response = self.connection.getresponse()
class ERP5TestReportHandler:
def __init__(self, url, suite_name):
# random test id
self.test_id = "%s-%X" % (
("%s" % datetime.date.today()).replace("-", ""),
randint(1, 10000000000000000),
)
self.connection_helper = ConnectionHelper(url)
self.suite_name = suite_name
def reportStart(self):
# report that test is running
print 'Starting test with id %s' % self.test_id
self.connection_helper.POST('TestResultModule_reportRunning', dict(
test_suite=self.suite_name,
test_report_id=self.test_id,
))
def reportFinished(self, out_file):
# make file parsable by erp5_test_results
out_file, success, failure, duration = self.processResult(out_file)
# XXX-Cedric : make correct display in test_result_module
tempcmd = tempfile.mkstemp()[1]
tempcmd2 = tempfile.mkstemp()[1]
tempout = tempfile.mkstemp()[1]
templog = tempfile.mkstemp()[1]
tl = open(templog, 'w')
tl.write(TB_SEP + '\n')
tl.write(out_file)
tl.write("----------------------------------------------------------------------\n")
tl.write('Ran 1 test in %.2fs\n' % duration)
if success:
tl.write('OK\n')
else:
tl.write('FAILED (failures=1)\n')
tl.write(TB_SEP + '\n')
tl.close()
open(tempcmd, 'w').write(""" %s""" % self.suite_name)
# create nice zip archive
tempzip = tempfile.mkstemp()[1]
# XXX-Cedric : support multiple tests
zip = zipfile.ZipFile(tempzip, 'w')
zip.write(tempout, '%s/001/stdout' % self.suite_name)
zip.write(templog, '%s/001/stderr' % self.suite_name)
zip.write(tempcmd, '%s/001/cmdline' % self.suite_name)
zip.close()
os.unlink(templog)
os.unlink(tempcmd)
os.unlink(tempout)
os.unlink(tempcmd2)
# post it to ERP5
self.connection_helper.POST('TestResultModule_reportCompleted',
dict(test_report_id=self.test_id), file_list=[('filepath', tempzip)])
os.unlink(tempzip)
def processResult(self, out_file):
file_content = out_file
sucess_amount = TEST_PASS_RE.search(file_content).group(1)
failure_amount = TEST_FAILURE_RE.search(file_content).group(1)
error_title_list = [re.compile('\s+').sub(' ', x).strip()
for x in TEST_ERROR_TITLE_RE.findall(file_content)]
duration = DURATION_RE.search(file_content).group(1)
detail = ''
for test_result in TEST_RESULT_RE.findall(file_content):
if TEST_ERROR_RESULT_RE.match(test_result):
detail += test_result
detail = IMAGE_RE.sub('', detail)
if detail:
detail = IMAGE_RE.sub('', detail)
detail = '''<html>
<head>
<style type="text/css">tr.status_failed { background-color:red };</style>
</head>
<body>%s</body>
</html>''' % detail
return detail, int(sucess_amount), int(failure_amount), float(duration)
#!/bin/sh
# BEWARE: This file is operated by slapgrid
# BEWARE: It will be overwritten automatically
%(xvfb_binary) :%(display) -screen %(display) 1024x768x24
\ No newline at end of file
##############################################################################
#
# Copyright (c) 2010 Vifib SARL and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 3
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
#############################################################################
from datetime import datetime
from erp5functionaltestreporthandler import ERP5TestReportHandler
from ERP5TypeFunctionalTestCase import TimeoutError
from time import sleep
import time
import os
import urllib2
import urlparse
from subprocess import Popen, PIPE
import signal
def run(suite_url, report_url, project, browser_binary):
suite_parsed = urlparse.urlparse(suite_url)
config = {
'suite_name': suite_parsed.path.split('/')[-1],
'base_url': "%s://%s%s" % (suite_parsed.scheme, suite_parsed.hostname,
'/'.join(suite_parsed.path.split('/')[:-1])),
'user': suite_parsed.username,
'password': suite_parsed.password,
}
test_url = assembleTestUrl(config['base_url'], config['suite_name'],
config['user'], config['password'])
# There is no test that can take more them 24 hours
timeout = 2.0 * 60 * 60
while True:
erp5_report = ERP5TestReportHandler(report_url,
project + '@' + config['suite_name'])
try:
try:
start = time.time()
print("Running test on: %s" % test_url)
process = Popen('%s "%s"' % (browser_binary, test_url),
stdout=PIPE,
stderr=PIPE,
shell=True,
close_fds=True)
erp5_report.reportStart()
while not isTestFinished(config['base_url']):
time.sleep(10)
print("Test not finished yet.")
if (time.time() - start) > float(timeout):
raise TimeoutError("Test took more than %s seconds" % timeout)
except TimeoutError:
continue
finally:
if process.pid:
os.kill(process.pid, signal.SIGTERM)
print("Test has finished and Firefox has been killed.")
erp5_report.reportFinished(getStatus(config['base_url']).encode("utf-8",
"replace"))
print("Test finished and report sent, sleeping.")
except urllib2.URLError, urlError:
print "Error: %s" % urlError.msg
sleep(3600)
def openUrl(url):
# Send Accept-Charset headers to activate the UnicodeConflictResolver
# (imitating firefox 3.5.9 here)
headers = { 'Accept-Charset' : 'ISO-8859-1,utf-8;q=0.7,*;q=0.7' }
request = urllib2.Request(url, headers=headers)
# Try to use long timeout, this is needed when there is many
# activities runing
try:
f = urllib2.urlopen(request, timeout=3600*4)
except TypeError:
f = urllib2.urlopen(request)
file_content = f.read()
f.close()
return file_content
def isTestFinished(url):
"""Fetch latest report. If report has been created less than 60 seconds ago,
it must be the current one.
Return true if test is finished, else return false.
"""
latest_report = openUrl('%s/portal_tests/TestTool_getLatestReportId/' % url)
if latest_report is '':
return False
latest_report_date = latest_report[7:]
time_delta = datetime.now() - \
datetime.strptime(latest_report_date, '%Y%m%d_%H%M%S' )
if time_delta.days is not 0:
return False
if time_delta.seconds < 120:
return True
return False
def getStatus(url):
try:
# Try 5 times.
for i in range(5):
try:
status = openUrl('%s/portal_tests/TestTool_getResults/' % (url))
break
except urllib2.URLError, urlError:
if i is 4: raise
print("Warning : %s while getting status" % urlError.msg)
except urllib2.HTTPError, e:
if e.msg == "No Content":
status = ""
else:
raise
return status
def assembleTestUrl(base_url, suite_name, user, password):
"""
Create the full url to the testrunner
"""
test_url = "%s/%s/core/TestRunner.html?test=../test_suite_html&"\
"resultsUrl=%s/postResults&auto=on&__ac_name=%s&__ac_password=%s" % (
base_url, suite_name, base_url, user, password)
return test_url
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment