ariya/phantomjs
1
1
Fork 0
mirror of https://github.com/ariya/phantomjs.git synced 2024-10-26 14:29:13 +03:00
Code Issues Projects Releases Wiki Activity

Overhaul run-tests.py.

Browse Source 
* Eliminate the global variables.
 * More thorough error detection in the HTTP(S) server.
 * Test directive comments must start with //! and the parser is more robust.
 * Handle child process stderr separately from stdout
   (this will make it possible to test more things).
 * Execute tests within each directory in alphabetical order. (#12439)
 * Support and use multi-level wildcards in TESTS.  This means we don't
   have to touch run-tests.py every time we add a subdirectory to
   module/ or standards/.
 * Handle HTTP server errors during a test more gracefully.  Errors caused by
   the client disconnecting in the middle of a query-response transaction are
   ignored.  Other errors are reported and the test suite exits cleanly.
 * Add HTTPS and POST support to test server.
 * Add ability to run PhantomJS under a debugger on a test case.
 * Use argparse for command line parsing.
 * Accept -v as short for --verbose.
 * Repeating -v means more verbosity; at -vvv, phantomjs command lines are
   printed.  (-vv is reserved for printing individual subtests, which will
   happen later.)
 * Fix bug where, with one or more -v, the pjs version number was printed
   twice.

Part of issue #13478 (test suite overhaul).
This commit is contained in:
Zack Weinberg 2015-08-13 10:55:48 -04:00
parent b483dd673a
commit 0cc40ddefe
8 changed files with 411 additions and 203 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
test/certs/https-snakeoil.crt Normal file
View File

@ -0,0 +1,18 @@
-----BEGIN CERTIFICATE-----
MIIC+zCCAeOgAwIBAgIJAJ7HwZBrgnLwMA0GCSqGSIb3DQEBCwUAMBQxEjAQBgNV
BAMMCWxvY2FsaG9zdDAeFw0xNTA4MTEyMjU4MTZaFw0yNTA4MTAyMjU4MTZaMBQx
EjAQBgNVBAMMCWxvY2FsaG9zdDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoC
ggEBANFha8c5JKjYrHc7BTqmuFSAxYsSKbUUa0k+0PFpjhj7Io/NOeHhxfdLJX/B
LVQXEDhvOlSTDBgC3RQkxCZJmMzKZjMDlj0cxY0esZtcqt0sRpwRvT+EBE9SlFu4
TWM2BQ6k5E4OIX/9aUk9HQ99pSjqmhu/7n76n/5DfqxGwkfVZengI1KwfezaB5+Q
wAvoS7tadROqTyynV1kd+OF9BJZwO1eR9lAiGc139J/BHegVcqdrI043oR+1vyTw
BFpodw4HYdJHNgo7DKAtmXoDAws5myqx2GcnVng1wyzu6LdM65nMV4/p5Y/Y6Ziy
RqeV1gVbtpxTcrLmWFnI8BRwFBUCAwEAAaNQME4wHQYDVR0OBBYEFPP1YOkZpJmE
x/W48Kwv2N1QC1oDMB8GA1UdIwQYMBaAFPP1YOkZpJmEx/W48Kwv2N1QC1oDMAwG
A1UdEwQFMAMBAf8wDQYJKoZIhvcNAQELBQADggEBAA1NxsrmKxGecAS6TEHNBqFZ
9NhV23kXY5sdv8zl7HUmzR+vIBumd9lkSZdOwAy5/hmj6ACReSJ9f2xpyi0fOtx5
WZ8Vcrg9Qjuy17qmGi66yL860yr0h6hltzCWCi7e26Eybawm3/9PmbNV3Hcwgxug
D+gv4LZLlyj4JI4lg/8RVXaNXqGBZ39YhRH0LFVjbYiFWUGqzfAT9YBoC67Ov8Yv
Bl1PoV3sJcagx67X6y8ru+gecc/OOXKJHxSidhjRqhKB6WOWIPfugsMOl1g2FMPv
tuPFsIQNSaln7V+ECeDOipJOSp9KAyM5fNcVjldd/e4V+qwcyoOijFywNfSK10M=
-----END CERTIFICATE-----

28
test/certs/https-snakeoil.key Normal file
View File

@ -0,0 +1,28 @@
-----BEGIN PRIVATE KEY-----
MIIEvAIBADANBgkqhkiG9w0BAQEFAASCBKYwggSiAgEAAoIBAQDRYWvHOSSo2Kx3
OwU6prhUgMWLEim1FGtJPtDxaY4Y+yKPzTnh4cX3SyV/wS1UFxA4bzpUkwwYAt0U
JMQmSZjMymYzA5Y9HMWNHrGbXKrdLEacEb0/hARPUpRbuE1jNgUOpORODiF//WlJ
PR0PfaUo6pobv+5++p/+Q36sRsJH1WXp4CNSsH3s2gefkMAL6Eu7WnUTqk8sp1dZ
HfjhfQSWcDtXkfZQIhnNd/SfwR3oFXKnayNON6Eftb8k8ARaaHcOB2HSRzYKOwyg
LZl6AwMLOZsqsdhnJ1Z4NcMs7ui3TOuZzFeP6eWP2OmYskanldYFW7acU3Ky5lhZ
yPAUcBQVAgMBAAECggEAOwI/w8fhAwz9niSuFpeB/57DDayywGveyKfBbygWegfc
97YZCAX/KvCswtKImdheI+mFAOzoTaQQ9mpeNYQsYhrwrpPmNZb0Pg9WcritFuQx
ii6drVbheBGH6kmI1dsVlcj25uCopE+g6pkkpYb9kwh7IjL3XiX4DUqsWpUej+ub
2iL/luW7nYHHIRqzOFgP3v/f29sFHNvYcgihphBMHtgb4VpeYQ/f7AC7k1bFYfA/
TmvfUcXdiPwJf0XICZOaLrT/6pigk0bRiLNn8npISu7Wlf4jF60bNAe4+krBVU4O
p8UjW99LiGKLDh8GpoudnzlnnngZ3SA5+bO7kwTjCQKBgQDvJwUShOWm2+5wJsr4
hWieTVDaZEDb+1WTe7DwtqRWNBWXchh8is9buWeXIe6+1WldBYYiQjgdggQCw8xG
IFFg1j1E6kPqk/kzrHYSsJ+/u8uaxypvvrVBhUqt5FduOxFojW2REX9W5n8HTdT4
32BGR4mGpuXzR+BsVK00QRgM+wKBgQDgIXtu6rbfx+mdXTFi6apWJoyu5bRWcrL2
mGoR+IjCk6NefcvBE33q54H/dk3+0Sxp6+uFo7lyKv4EL3lozQO2oga6sp2LOIEK
DUo+KQVOmntCNrjuN/PbjSu2s1j5QDnLNR9VvXGiYBWdpZ7k3YzoKJ1I4ZyB3kGs
H/lCXv52LwKBgER1HvaWJEcHXdGsyR0q0y+9Yg+h8w8FexGkrpm5LoGely+q8Wd1
NLZE9GpGxFjMLkT6d9MGsZmAxjUkZy0Lwz+9E/zOMnLLuOIZ1BK1jIUN9NJxgKxM
IwaGaUItwvlC31DWay7Dm3f8sxAcL4KuLpjvkWaCEAD76joYYxw6JfBRAoGADMe7
+xolLWN/3bpHq6U5UkpGcV6lxtwpekg8nCO44Kd8hFHWAX90CaYD0qZTUjlpN+z8
9BTe6TSsYV63pJM0KADbM2Al/Z9ONF2Hoz3BkLbcWm02ZFcKb7WADZ3yb9wKr5yq
2b/AsAqckO21vsUnWMGgHlzHCNy8j+0O0IsMJX8CgYAORhyGaU7x5t4kEvqBNIan
mOzuB0b5nYV9mHxmyFhOsa8LeM25SA4n1rFpTb8h6vmZF1y9+4Zy4uNfCR2wXg0v
I51qtZ8npbIksYvNqvHaTPg8ZBcFK5mHr3TDxXJCcc0ylzM98ze08D+qKr0joX4w
KlqN6KjGmYfb+RHehLk9sw==
-----END PRIVATE KEY-----

2
test/module/webpage/local-urls-disabled-iframe.js
View File

@ -1,4 +1,4 @@
// phantomjs: --web-security=no --local-url-access=no
//! phantomjs: --web-security=no --local-url-access=no
var assert = require("../../assert");
var p = require("webpage").create();

2
test/module/webpage/local-urls-disabled.js
View File

@ -1,4 +1,4 @@
// phantomjs: --local-url-access=no
//! phantomjs: --local-url-access=no
var assert = require("../../assert");
var p = require("webpage").create();

2
test/module/webpage/local-urls-enabled-iframe.js
View File

@ -1,4 +1,4 @@
// phantomjs: --web-security=no --local-url-access=yes
//! phantomjs: --web-security=no --local-url-access=yes
var assert = require("../../assert");
var p = require("webpage").create();

2
test/module/webpage/local-urls-enabled.js
View File

@ -1,4 +1,4 @@
// phantomjs: --local-url-access=yes
//! phantomjs: --local-url-access=yes
var assert = require("../../assert");
var p = require("webpage").create();

557
test/run-tests.py
View File

@ -1,64 +1,154 @@
#!/usr/bin/env python
import cStringIO as StringIO
import argparse
import errno
import glob
import imp
import optparse
import os
import posixpath
import shlex
import SimpleHTTPServer
import SocketServer
import socket
import SocketServer
import ssl
import string
import cStringIO as StringIO
import subprocess
import sys
import threading
import time
import traceback
import urllib
TIMEOUT = 10 # Maximum duration of PhantomJS execution (in seconds)
HTTP_PORT = 9180
http_running = False
# All files matching one of these glob patterns will be run as tests.
TESTS = [
'basics/*.js',
'module/system/*.js',
'module/webpage/*.js',
'module/cookiejar/*.js',
'standards/javascript/*.js',
'standards/console/*.js',
'module/*/*.js',
'standards/*/*.js',
'regression/*.js',
'run-tests.js'
]
# This should be in the standard library somewhere, but as far as I
# can tell, isn't.
def import_file_as_module(path):
# All Python response hooks, no matter how deep below www_path,
# are treated as direct children of the fake "test_www" package.
if 'test_www' not in sys.modules:
imp.load_source('test_www', www_path + '/__init__.py')
TIMEOUT = 20 # Maximum duration of PhantomJS execution (in seconds).
# There's currently no way to adjust this on a per-test
# basis, so it has to be large.
HTTP_PORT = 9180 # These are currently hardwired into every test that
HTTPS_PORT = 9181 # uses the test servers.
# create_default_context and SSLContext were only added in 2.7.9,
# which is newer than the python2 that ships with OSX :-(
# The fallback tries to mimic what create_default_context(CLIENT_AUTH)
# does. Security obviously isn't important in itself for a test
# server, but making sure the PJS client can talk to a server
# configured according to modern TLS best practices _is_ important.
# Unfortunately, there is no way to set things like OP_NO_SSL2 or
# OP_CIPHER_SERVER_PREFERENCE prior to 2.7.9.
CIPHERLIST_2_7_9 = (
'ECDH+AESGCM:DH+AESGCM:ECDH+AES256:DH+AES256:ECDH+AES128:DH+AES:ECDH+HIGH:'
'DH+HIGH:ECDH+3DES:DH+3DES:RSA+AESGCM:RSA+AES:RSA+HIGH:RSA+3DES:!aNULL:'
'!eNULL:!MD5:!DSS:!RC4'
)
def wrap_socket_ssl(sock, base_path):
crtfile = os.path.join(base_path, 'certs/https-snakeoil.crt')
keyfile = os.path.join(base_path, 'certs/https-snakeoil.key')
tr = string.maketrans('-./%', '____')
modname = 'test_www.' + path.translate(tr)
try:
return sys.modules[modname]
except KeyError:
return imp.load_source(modname, path)
ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ctx.load_cert_chain(crtfile, keyfile)
return ctx.wrap_socket(sock, server_side=True)
except AttributeError:
return ssl.wrap_socket(sock,
keyfile=keyfile,
certfile=crtfile,
server_side=True,
ciphers=CIPHERLIST_2_7_9)
# This should be in the standard library somewhere, but as far as I
# can tell, it isn't.
class ResponseHookImporter(object):
def __init__(self, www_path):
# All Python response hooks, no matter how deep below www_path,
# are treated as direct children of the fake "test_www" package.
if 'test_www' not in sys.modules:
imp.load_source('test_www', www_path + '/__init__.py')
self.tr = string.maketrans('-./%', '____')
def __call__(self, path):
modname = 'test_www.' + path.translate(self.tr)
try:
return sys.modules[modname]
except KeyError:
return imp.load_source(modname, path)
# This should also be in the standard library somewhere, and definitely
# isn't.
# FIXME: Use non-blocking I/O and select on Unix for efficiency.
# FIXME: Can hang forever if the subprocess closes its stdout but doesn't exit.
def record_process_output(proc, verbose, timeout):
def read_thread(linebuf, fp):
while True:
line = fp.readline().rstrip()
if not line: break # EOF
line = line.rstrip()
if line:
linebuf.append(line)
if verbose:
sys.stdout.write(line + '\n')
stdout = []
stderr = []
sothrd = threading.Thread(target=read_thread, args=(stdout, proc.stdout))
sethrd = threading.Thread(target=read_thread, args=(stderr, proc.stderr))
timed_out = False
sothrd.start()
sethrd.start()
sothrd.join(timeout)
if sothrd.is_alive():
timed_out = True
proc.terminate()
sothrd.join()
sethrd.join()
proc.wait()
if timed_out:
stderr.append(" TIMEOUT | Process terminated after {} seconds."
.format(timeout))
return proc.returncode, stdout, stderr
class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler, object):
def __init__(self, *args, **kwargs):
self._cached_untranslated_path = None
self._cached_translated_path = None
self.postdata = None
super(FileHandler, self).__init__(*args, **kwargs)
# silent, do not pollute stdout nor stderr.
def log_message(self, format, *args):
return
# accept POSTs, read the postdata and stash it in an instance variable,
# then forward to do_GET; handle_request hooks can vary their behavior
# based on the presence of postdata and/or the command verb.
def do_POST(self):
try:
ln = int(self.headers.get('content-length'))
except TypeError, ValueError:
self.send_response(400, 'Bad Request')
self.send_header('Content-Type', 'text/plain')
self.end_headers()
self.wfile.write("No or invalid Content-Length in POST (%r)"
% self.headers.get('content-length'))
return
self.postdata = self.rfile.read(ln)
self.do_GET()
# allow provision of a .py file that will be interpreted to
# produce the response.
def send_head(self):
@ -77,19 +167,11 @@ class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler, object):
py = path + '.py'
if os.path.exists(py):
try:
mod = import_file_as_module(py)
mod = self.get_response_hook(py)
return mod.handle_request(self)
except:
import cgitb
buf = StringIO.StringIO()
cgitb.Hook(file=buf).handle()
buf = buf.getvalue()
self.send_response(500, 'Internal Server Error')
self.send_header('Content-Type', 'text/html')
self.send_header('Content-Length', str(len(buf)))
self.end_headers()
return StringIO.StringIO(buf)
self.send_error(500, 'Internal Server Error in '+py)
raise
self.send_error(404, 'File not found')
return None
@ -134,7 +216,7 @@ class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler, object):
# Now resolve the normalized, clamped path relative to the www/
# directory, according to local OS conventions.
path = os.path.normpath(os.path.join(www_path, *path.split('/')))
path = os.path.normpath(os.path.join(self.www_path, *path.split('/')))
if trailing_slash:
# it must be a '/' even on Windows
path += '/'
@ -143,189 +225,268 @@ class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler, object):
self._cached_translated_path = path
return path
# This is how you are officially supposed to set SO_REUSEADDR per
# https://docs.python.org/2/library/socketserver.html#SocketServer.BaseServer.allow_reuse_address
class TCPServer(SocketServer.TCPServer):
class TCPServer(SocketServer.ThreadingMixIn, SocketServer.TCPServer):
# This is how you are officially supposed to set SO_REUSEADDR per
# https://docs.python.org/2/library/socketserver.html#SocketServer.BaseServer.allow_reuse_address
allow_reuse_address = True
def run_httpd():
global http_running
handler = FileHandler
handler.extensions_map.update({
'.htm': 'text/html',
'.html': 'text/html',
'.css': 'text/css',
'.js': 'application/javascript',
'.json': 'application/json'
})
try:
httpd = TCPServer(('', HTTP_PORT), handler)
while http_running:
httpd.handle_request()
except socket.error as e:
print 'Fatal error: unable to launch a test server at port', HTTP_PORT
print str(e)
http_running = False
sys.exit(1)
return
def __init__(self, port, use_ssl, handler, base_path, signal_error):
SocketServer.TCPServer.__init__(self, ('localhost', port), handler)
if use_ssl:
self.socket = wrap_socket_ssl(self.socket, base_path)
self._signal_error = signal_error
def handle_error(self, request, client_address):
# Ignore errors which can occur naturally if the client
# disconnects in the middle of a request. EPIPE and
# ECONNRESET *should* be the only such error codes
# (according to the OSX manpage for send()).
_, exval, _ = sys.exc_info()
if getattr(exval, 'errno', None) in (errno.EPIPE, errno.ECONNRESET):
return
def setup_server():
global http_running
# Otherwise, report the error to the test runner.
self._signal_error(sys.exc_info())
http_running = True
httpd_thread = threading.Thread(target=run_httpd)
httpd_thread.start()
class HTTPTestServer(object):
def __init__(self, base_path, signal_error):
self.httpd = None
self.httpsd = None
self.base_path = base_path
self.www_path = os.path.join(base_path, 'www')
self.signal_error = signal_error
time.sleep(2)
if http_running:
if options.verbose:
print 'serving at port', HTTP_PORT
def __enter__(self):
handler = FileHandler
handler.extensions_map.update({
'.htm': 'text/html',
'.html': 'text/html',
'.css': 'text/css',
'.js': 'application/javascript',
'.json': 'application/json'
})
handler.www_path = self.www_path
handler.get_response_hook = ResponseHookImporter(self.www_path)
self.httpd = TCPServer(HTTP_PORT, False,
handler, self.base_path, self.signal_error)
httpd_thread = threading.Thread(target=self.httpd.serve_forever)
httpd_thread.daemon = True
httpd_thread.start()
def terminate_server():
global http_running
http_running = False
self.httpsd = TCPServer(HTTPS_PORT, True,
handler, self.base_path, self.signal_error)
httpsd_thread = threading.Thread(target=self.httpsd.serve_forever)
httpsd_thread.daemon = True
httpsd_thread.start()
# ping the server once to trigger the check for http_running (after every request)
urllib.urlopen('http://localhost:{0}'.format(HTTP_PORT))
return self
def __exit__(self, *dontcare):
self.httpd.shutdown()
self.httpsd.shutdown()
class TestRunner(object):
def __init__(self, base_path, phantomjs_exe, options):
self.base_path = base_path
self.cert_path = os.path.join(base_path, 'certs')
self.phantomjs_exe = phantomjs_exe
self.verbose = options.verbose
self.debugger = options.debugger
self.to_run = options.to_run
self.http_server_err = None
def signal_server_error(self, exc_info):
self.http_server_err = exc_info
def get_base_command(self, debugger):
if debugger is None:
return [self.phantomjs_exe]
elif debugger == "gdb":
return ["gdb", "--args", self.phantomjs_exe]
elif debugger == "lldb":
return ["lldb", "--", self.phantomjs_exe]
else:
raise RuntimeError("Don't know how to invoke " + self.debugger)
def run_phantomjs(self, script, script_args=[], pjs_args=[], silent=False):
verbose = self.verbose
debugger = self.debugger
if silent:
verbose = False
debugger = None
output = []
command = self.get_base_command(debugger)
command.extend(pjs_args)
command.append(script)
if verbose:
command.append('--verbose={}'.format(verbose))
command.extend(script_args)
if verbose >= 3:
sys.stdout.write("## running {}\n".format(" ".join(command)))
if debugger:
subprocess.call(command)
return 0, [], []
else:
proc = subprocess.Popen(command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
return record_process_output(proc, verbose, TIMEOUT)
def run_test(self, script, name):
script_args = []
pjs_args = []
use_snakeoil = True
# Parse any directives at the top of the script.
try:
with open(script, "rt") as s:
for line in s:
if not line.startswith("//!"):
break
tokens = shlex.split(line[3:], comments=True)
for i in range(len(tokens)):
tok = tokens[i]
if tok == "no-snakeoil":
use_snakeoil = False
elif tok == "phantomjs:":
if i+1 == len(tokens):
raise ValueError("phantomjs: directive requires"
"at least one argument")
pjs_args.extend(tokens[(i+1):])
break
elif tok == "script:":
if i+1 == len(tokens):
raise ValueError("script: directive requires"
"at least one argument")
script_args.extend(tokens[(i+1):])
break
else:
raise ValueError("unrecognized directive: " + tok)
except OSError as e:
sys.stdout.write('{} ({}): {}\n'
.format(name, e.filename, e.strerror))
return 1
except Exception as e:
sys.stdout.write('{} ({}): {}\n'
.format(name, script, str(e)))
return 1
if use_snakeoil:
pjs_args.insert(0, '--ssl-certificates-path=' + self.cert_path)
sys.stdout.write(name + ":\n")
self.http_server_err = None
returncode, out, err = self.run_phantomjs(script, script_args, pjs_args)
if returncode != 0:
if not self.verbose:
if out:
sys.stdout.write("\n".join(out))
sys.stdout.write("\n")
if err:
sys.stdout.write("\n".join(err))
sys.stdout.write("\n")
if self.http_server_err is not None:
ty, val, tb = self.http_server_err
sys.stdout.write(" ERROR | httpd: {}\n".format(
traceback.format_exception_only(ty, val)[-1]))
if self.verbose:
for line in traceback.format_tb(tb, 5):
sys.stdout.write("## " + line)
return 1
return returncode
def run_tests(self):
start = time.time()
result = 0
any_executed = False
base = self.base_path
nlen = len(base) + 1
for test_group in TESTS:
test_glob = os.path.join(base, test_group)
for test_script in sorted(glob.glob(test_glob)):
tname = os.path.splitext(test_script)[0][nlen:]
if self.to_run:
for to_run in self.to_run:
if to_run in tname:
break
else:
continue
any_executed = True
ret = self.run_test(test_script, tname)
if ret != 0:
sys.stdout.write('The test {} FAILED\n\n'.format(tname))
result = 1
if not any_executed:
sys.stdout.write("All tests skipped.\n")
return 1
if result == 0:
sys.stdout.write("\nAll tests successful. "
"Total time: {:.3f} seconds.\n"
.format(time.time() - start))
return result
def init():
global base_path, www_path, phantomjs_exe, options
base_path = os.path.dirname(os.path.abspath(__file__))
base_path = os.path.normpath(os.path.dirname(os.path.abspath(__file__)))
phantomjs_exe = os.path.normpath(base_path + '/../bin/phantomjs')
if sys.platform in ('win32', 'cygwin'):
phantomjs_exe += '.exe'
if not os.path.isfile(phantomjs_exe):
print 'Could not locate ' + phantomjs_exe
sys.stdout.write("{} is unavailable, cannot run tests.\n"
.format(phantomjs_exe))
sys.exit(1)
www_path = os.path.join(base_path, 'www')
parser = optparse.OptionParser(
usage='%prog [options] [tests to run...]',
description='Run PhantomJS tests.'
)
parser.add_option('--verbose', action='store_true', default=False, help='Show a verbose log')
(options, args) = parser.parse_args(sys.argv[1:])
options.to_run = args
parser = argparse.ArgumentParser(description='Run PhantomJS tests.')
parser.add_argument('-v', '--verbose', action='count',
help='Increase verbosity of logs (repeat for more)')
parser.add_argument('to_run', nargs='*', metavar='test',
help='tests to run (default: all of them)')
parser.add_argument('--debugger', default=None,
help="Run PhantomJS under DEBUGGER")
options = parser.parse_args()
runner = TestRunner(base_path, phantomjs_exe, options)
if options.verbose:
returncode, version = run_phantomjs('--version')
print 'Checking PhantomJS version %s' % version
rc, ver, err = runner.run_phantomjs('--version', silent=True)
if rc != 0 or len(ver) != 1 or len(err) != 0:
sys.stdout.write(" ERROR | Version check failed\n")
for l in ver: sys.stdout.write("## {}\n".format(l))
for l in err: sys.stdout.write("## {}\n".format(l))
sys.stdout.write("## exit {}".format(rc))
sys.exit(1)
sys.stdout.write(" INFO | Testing PhantomJS {}\n".format(ver[0]))
def run_phantomjs(script, script_args=[], pjs_args=[]):
output = []
command = [phantomjs_exe]
command.extend(pjs_args)
command.append(script)
command.extend(script_args)
process = subprocess.Popen(command, stderr=subprocess.STDOUT, stdout=subprocess.PIPE)
def runner():
while True:
line = process.stdout.readline()
if line != '':
output.append(line.rstrip())
if options.verbose:
print '%s' % line.rstrip()
else:
break
thread = threading.Thread(target=runner)
thread.start()
thread.join(TIMEOUT)
if thread.is_alive():
print 'Process is running more than ' + str(TIMEOUT) + ' seconds. Terminating...'
process.terminate()
thread.join()
process.wait()
return process.returncode, '\n'.join(output)
def run_test(script, name):
script_args = []
pjs_args = []
if options.verbose:
script_args.append('--verbose')
return runner
def main():
runner = init()
try:
with open(script, "rt") as s:
p_prefix = "// phantomjs: "
s_prefix = "// script: "
for line in s:
if line.startswith(p_prefix):
pjs_args.extend(shlex.split(line[len(p_prefix):]))
if line.startswith(s_prefix):
script_args.extend(shlex.split(line[len(s_prefix):]))
if not line.startswith("//"):
break
except OSError as e:
print '%s: %s: %s' % (name, e.filename, e.strerror)
return 1
with HTTPTestServer(runner.base_path, runner.signal_server_error):
sys.exit(runner.run_tests())
print '%s:' % name
returncode, output = run_phantomjs(script, script_args, pjs_args)
if returncode != 0:
if not options.verbose:
print '%s' % output
return 1
except Exception as e:
ty, val, tb = sys.exc_info()
sys.stdout.write(" ERROR | {}\n".format(
traceback.format_exception_only(ty, val)[-1]))
if runner.verbose:
for line in traceback.format_tb(tb, 5):
sys.stdout.write("## " + line)
return 0
sys.exit(1)
def run_tests():
setup_server()
if not http_running:
return 1
start = time.time()
if options.verbose:
print 'Starting the tests...'
result = 0
any_executed = False
for test_group in TESTS:
test_group_name = os.path.dirname(test_group)
test_glob = os.path.normpath(base_path + '/' + test_group)
if options.verbose:
print
print 'Test group: %s...' % test_group_name
for test_script in glob.glob(test_glob):
tname = test_group_name + '/' + os.path.basename(test_script)
if options.to_run:
for to_run in options.to_run:
if to_run in tname:
break
else:
if options.verbose:
print "%s: skipped" % tname
continue
any_executed = True
ret = run_test(test_script, tname)
if ret != 0:
print 'The test %s FAILED' % tname
print
result = 1
if not any_executed:
result = 1
print
print 'ALL TESTS SKIPPED.'
if result == 0:
print
print 'No failure. Finished in %d seconds.' % (time.time() - start)
terminate_server()
return result
init()
sys.exit(run_tests())
main()

3
test/www/echo.py
View File

@ -17,7 +17,8 @@ def handle_request(req):
params = url.params,
query = url.query,
fragment = url.fragment,
headers = headers
headers = headers,
postdata = req.postdata
)
body = json.dumps(d, indent=2) + '\n'