phantomjs/test/run-tests.py

#!/usr/bin/env python

import cStringIO as StringIO
import glob
import imp
import optparse
import os
import posixpath
import shlex
import SimpleHTTPServer
import SocketServer
import socket
import string
import subprocess
import sys
import threading
import time
import urllib

TIMEOUT = 10  # Maximum duration of PhantomJS execution (in seconds)

HTTP_PORT = 9180
http_running = False

TESTS = [
    'basics/*.js',
    'module/system/*.js',
    'module/webpage/*.js',
    'module/cookiejar/*.js',
    'standards/javascript/*.js',
    'standards/console/*.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')

    tr = string.maketrans('-./%', '____')
    modname = 'test_www.' + path.translate(tr)
    try:
        return sys.modules[modname]
    except KeyError:
        return imp.load_source(modname, path)

class FileHandler(SimpleHTTPServer.SimpleHTTPRequestHandler, object):

    def __init__(self, *args, **kwargs):
        self._cached_untranslated_path = None
        self._cached_translated_path = None
        super(FileHandler, self).__init__(*args, **kwargs)

    # silent, do not pollute stdout nor stderr.
    def log_message(self, format, *args):
        return

    # allow provision of a .py file that will be interpreted to
    # produce the response.
    def send_head(self):
        path = self.translate_path(self.path)

        # do not allow direct references to .py(c) files,
        # or indirect references to __init__.py
        if (path.endswith('.py') or path.endswith('.pyc') or
            path.endswith('__init__')):
            self.send_error(404, 'File not found')
            return None

        if os.path.exists(path):
            return super(FileHandler, self).send_head()

        py = path + '.py'
        if os.path.exists(py):
            try:
                mod = import_file_as_module(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(404, 'File not found')
        return None

    # modified version of SimpleHTTPRequestHandler's translate_path
    # to resolve the URL relative to the www/ directory
    # (e.g. /foo -> test/www/foo)
    def translate_path(self, path):

        # Cache for efficiency, since our send_head calls this and
        # then, in the normal case, the parent class's send_head
        # immediately calls it again.
        if (self._cached_translated_path is not None and
            self._cached_untranslated_path == path):
            return self._cached_translated_path

        orig_path = path

        # Strip query string and/or fragment, if present.
        x = path.find('?')
        if x != -1: path = path[:x]
        x = path.find('#')
        if x != -1: path = path[:x]

        # Ensure consistent encoding of special characters, then
        # lowercase everything so that the tests behave consistently
        # whether or not the local filesystem is case-sensitive.
        path = urllib.quote(urllib.unquote(path)).lower()

        # Prevent access to files outside www/.
        # At this point we want specifically POSIX-like treatment of 'path'
        # because it is still a URL component and not a filesystem path.
        # SimpleHTTPRequestHandler.send_head() expects us to preserve the
        # distinction between paths with and without a trailing slash, but
        # posixpath.normpath() discards that distinction.
        trailing_slash = path.endswith('/')
        path = posixpath.normpath(path)
        while path.startswith('/'):
            path = path[1:]
        while path.startswith('../'):
            path = path[3:]

        # 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('/')))
        if trailing_slash:
            # it must be a '/' even on Windows
            path += '/'

        self._cached_untranslated_path = orig_path
        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):
    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 setup_server():
    global http_running

    http_running = True
    httpd_thread = threading.Thread(target=run_httpd)
    httpd_thread.start()

    time.sleep(2)
    if http_running:
        if options.verbose:
            print 'serving at port', HTTP_PORT


def terminate_server():
    global http_running
    http_running = False

    # ping the server once to trigger the check for http_running (after every request)
    urllib.urlopen('http://localhost:{0}'.format(HTTP_PORT))


def init():
    global base_path, www_path, phantomjs_exe, options

    base_path = os.path.dirname(os.path.abspath(__file__))
    phantomjs_exe = os.path.normpath(base_path + '/../bin/phantomjs')
    if not os.path.isfile(phantomjs_exe):
        print 'Could not locate ' + 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

    if options.verbose:
        returncode, version = run_phantomjs('--version')
        print 'Checking PhantomJS version %s' % version


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')

    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

    print '%s:' % name
    returncode, output = run_phantomjs(script, script_args, pjs_args)
    if returncode != 0:
        if not options.verbose:
            print '%s' % output
        return 1

    return 0


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())