third_party/web_platform_tests/tools/wptrunner/test/test.py

import ConfigParser
import argparse
import json
import os
import sys
import tempfile
import threading
import time
from StringIO import StringIO

from mozlog import structuredlog, reader
from mozlog.handlers import BaseHandler, StreamHandler, StatusHandler
from mozlog.formatters import MachFormatter
from wptrunner import wptcommandline, wptrunner

here = os.path.abspath(os.path.dirname(__file__))

def setup_wptrunner_logging(logger):
    structuredlog.set_default_logger(logger)
    wptrunner.logger = logger
    wptrunner.wptlogging.setup_stdlib_logger()

class ResultHandler(BaseHandler):
    def __init__(self, verbose=False, logger=None):
        self.inner = StreamHandler(sys.stdout, MachFormatter())
        BaseHandler.__init__(self, self.inner)
        self.product = None
        self.verbose = verbose
        self.logger = logger

        self.register_message_handlers("wptrunner-test", {"set-product": self.set_product})

    def set_product(self, product):
        self.product = product

    def __call__(self, data):
        if self.product is not None and data["action"] in ["suite_start", "suite_end"]:
            # Hack: mozlog sets some internal state to prevent multiple suite_start or
            # suite_end messages. We actually want that here (one from the metaharness
            # and one from the individual test type harness), so override that internal
            # state (a better solution might be to not share loggers, but this works well
            # enough)
            self.logger._state.suite_started = True
            return

        if (not self.verbose and
            (data["action"] == "process_output" or
             data["action"] == "log" and data["level"] not in ["error", "critical"])):
            return

        if "test" in data:
            data = data.copy()
            data["test"] = "%s: %s" % (self.product, data["test"])

        return self.inner(data)

def test_settings():
    return {
        "include": "_test",
        "manifest-update": "",
        "no-capture-stdio": ""
    }

def read_config():
    parser = ConfigParser.ConfigParser()
    parser.read("test.cfg")

    rv = {"general":{},
          "products":{}}

    rv["general"].update(dict(parser.items("general")))

    # This only allows one product per whatever for now
    for product in parser.sections():
        if product != "general":
            dest = rv["products"][product] = {}
            for key, value in parser.items(product):
                rv["products"][product][key] = value

    return rv

def run_tests(product, kwargs):
    kwargs["test_paths"]["/_test/"] = {"tests_path": os.path.join(here, "testdata"),
                                       "metadata_path": os.path.join(here, "metadata")}

    wptrunner.run_tests(**kwargs)

def settings_to_argv(settings):
    rv = []
    for name, value in settings.items():
        key = "--%s" % name
        if not value:
            rv.append(key)
        elif isinstance(value, list):
            for item in value:
                rv.extend([key, item])
        else:
            rv.extend([key, value])
    return rv

def set_from_args(settings, args):
    if args.test:
        settings["include"] = args.test
    if args.tags:
        settings["tags"] = args.tags

def run(config, args):
    logger = structuredlog.StructuredLogger("web-platform-tests")
    logger.add_handler(ResultHandler(logger=logger, verbose=args.verbose))
    setup_wptrunner_logging(logger)

    parser = wptcommandline.create_parser()

    logger.suite_start(tests=[])

    for product, product_settings in config["products"].items():
        if args.product and product not in args.product:
            continue

        settings = test_settings()
        settings.update(config["general"])
        settings.update(product_settings)
        settings["product"] = product
        set_from_args(settings, args)

        kwargs = vars(parser.parse_args(settings_to_argv(settings)))
        wptcommandline.check_args(kwargs)

        logger.send_message("wptrunner-test", "set-product", product)

        run_tests(product, kwargs)

    logger.send_message("wptrunner-test", "set-product", None)
    logger.suite_end()

def get_parser():
    parser = argparse.ArgumentParser()
    parser.add_argument("-v", "--verbose", action="store_true", default=False,
                        help="verbose log output")
    parser.add_argument("--product", action="append",
                        help="Specific product to include in test run")
    parser.add_argument("--pdb", action="store_true",
                        help="Invoke pdb on uncaught exception")
    parser.add_argument("--tag", action="append", dest="tags",
                        help="tags to select tests")
    parser.add_argument("test", nargs="*",
                        help="Specific tests to include in test run")
    return parser

def main():
    config = read_config()

    args = get_parser().parse_args()

    try:
        run(config, args)
    except Exception:
        if args.pdb:
            import pdb, traceback
            print traceback.format_exc()
            pdb.post_mortem()
        else:
            raise

if __name__ == "__main__":
    main()