| 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.iteritems(): |
| 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"].iteritems(): |
| 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() |