testing/scripts/wpt_common.py - chromium/src.git - Git at Google

 # Copyright 2020 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.

 import base64
 import json
 import os
 import shutil
 import sys

 import common

 BLINK_TOOLS_DIR = os.path.join(common.SRC_DIR, 'third_party', 'blink', 'tools')
 WEB_TESTS_DIR = os.path.join(BLINK_TOOLS_DIR, os.pardir, 'web_tests')

 if BLINK_TOOLS_DIR not in sys.path:
     sys.path.append(BLINK_TOOLS_DIR)

 from blinkpy.common.host import Host
 from blinkpy.web_tests.models import test_failures

 class BaseWptScriptAdapter(common.BaseIsolatedScriptArgsAdapter):
     """The base class for script adapters that use wptrunner to execute web
     platform tests. This contains any code shared between these scripts, such
     as integrating output with the results viewer. Subclasses contain other
     (usually platform-specific) logic."""

     def __init__(self):
         super(BaseWptScriptAdapter, self).__init__()
         host = Host()
         self.port = host.port_factory.get()

     def generate_test_output_args(self, output):
         return ['--log-chromium', output]

     def generate_sharding_args(self, total_shards, shard_index):
         return ['--total-chunks=%d' % total_shards,
                 # shard_index is 0-based but WPT's this-chunk to be 1-based
                 '--this-chunk=%d' % (shard_index + 1),
                 # The default sharding strategy is to shard by directory. But
                 # we want to hash each test to determine which shard runs it.
                 # This allows running individual directories that have few
                 # tests across many shards.
                 '--chunk-type=hash']

     def do_post_test_run_tasks(self):
         # Move json results into layout-test-results directory
         results_dir = os.path.dirname(self.options.isolated_script_test_output)
         layout_test_results = os.path.join(results_dir, 'layout-test-results')
         if os.path.exists(layout_test_results):
             shutil.rmtree(layout_test_results)
         os.mkdir(layout_test_results)

         # Perform post-processing of wptrunner output
         self.process_wptrunner_output()

         shutil.copyfile(self.options.isolated_script_test_output,
                         os.path.join(layout_test_results, 'full_results.json'))
         # create full_results_jsonp.js file which is used to
         # load results into the results viewer
         with open(self.options.isolated_script_test_output, 'r') \
             as full_results, \
             open(os.path.join(
                 layout_test_results, 'full_results_jsonp.js'), 'w') \
                 as json_js:
             json_js.write('ADD_FULL_RESULTS(%s);' % full_results.read())
         # copy layout test results viewer to layout-test-results directory
         shutil.copyfile(
             os.path.join(WEB_TESTS_DIR, 'fast', 'harness', 'results.html'),
             os.path.join(layout_test_results, 'results.html'))

     def process_wptrunner_output(self):
         """Post-process the output generated by wptrunner.

         This output contains a single large json file containing the raw content
         or artifacts which need to be extracted into their own files and removed
         from the json file (to avoid duplication)."""
         output_json = json.load(
             open(self.options.isolated_script_test_output, "r"))
         test_json = output_json["tests"]
         results_dir = os.path.dirname(self.options.isolated_script_test_output)
         self._process_test_leaves(results_dir, output_json["path_delimiter"],
                                   test_json, "")
         # Write output_json back to the same file after modifying it in memory
         with open(self.options.isolated_script_test_output, "w") as output_file:
             json.dump(output_json, output_file)

     def _process_test_leaves(self, results_dir, delim, root_node, path_so_far):
         """Finds and processes each test leaf below the specified root.

         This will recursively traverse the trie of results in the json output,
         keeping track of the path to each test and identifying leaves by the
         presence of certain attributes.

         Args:
             results_dir: str path to the dir that results are stored
             delim: str delimiter to be used for test names
             root_node: dict representing the root of the trie we're currently
                 looking at
             path_so_far: str the path to the current root_node in the trie
         """
         if "actual" in root_node:
             # Found a leaf, process it
             if "artifacts" not in root_node:
                 return
             log_artifact = root_node["artifacts"].pop("log", None)
             if log_artifact:
                 artifact_subpath = self._write_log_artifact(
                     test_failures.FILENAME_SUFFIX_ACTUAL,
                     results_dir, path_so_far, log_artifact)
                 root_node["artifacts"]["actual_text"] = [artifact_subpath]

             screenshot_artifact = root_node["artifacts"].pop("screenshots",
                                                              None)
             if screenshot_artifact:
                 screenshot_paths_dict = self._write_screenshot_artifact(
                     results_dir, path_so_far, screenshot_artifact)
                 for screenshot_key, path in screenshot_paths_dict.items():
                     root_node["artifacts"][screenshot_key] = [path]

             crashlog_artifact = root_node["artifacts"].pop("wpt_crash_log",
                                                            None)
             if crashlog_artifact:
                 artifact_subpath = self._write_log_artifact(
                     test_failures.FILENAME_SUFFIX_CRASH_LOG,
                     results_dir, path_so_far, crashlog_artifact)

             return

         # We're not at a leaf node, continue traversing the trie.
         for key in root_node:
             # Append the key to the path, separated by the delimiter. However if
             # the path is empty, skip the delimiter to avoid a leading slash in
             # the path.
             new_path = path_so_far + delim + key if path_so_far else key
             self._process_test_leaves(results_dir, delim, root_node[key],
                                       new_path)

     def _write_log_artifact(self, suffix, results_dir, test_name, log_artifact):
         """Writes a log artifact to disk.

         A log artifact contains some form of output for a test. It is written to
         a txt file with a suffix generated from the log type.

         Args:
             suffix: str suffix of the artifact to write, e.g.
                 test_failures.FILENAME_SUFFIX_ACTUAL
             results_dir: str path to the directory that results live in
             test_name: str name of the test that this artifact is for
             log_artifact: list of strings, the log entries for this test from
                 the json output.

         Returns:
             string path to the artifact file that the log was written to,
             relative to the directory that the original output is located.
         """
         log_artifact_sub_path = (
             os.path.join("layout-test-results",
                          self.port.output_filename(test_name, suffix, ".txt"))
         )
         log_artifact_full_path = os.path.join(results_dir,
                                               log_artifact_sub_path)
         if not os.path.exists(os.path.dirname(log_artifact_full_path)):
             os.makedirs(os.path.dirname(log_artifact_full_path))
         with open(log_artifact_full_path, "w") as artifact_file:
             artifact_file.write("\n".join(log_artifact).encode("utf-8"))

         return log_artifact_sub_path

     def _write_screenshot_artifact(self, results_dir, test_name,
                                    screenshot_artifact):
         """Write screenshot artifact to disk.

         The screenshot artifact is a list of strings, each of which has the
         format <url>:<base64-encoded PNG>. Each url-png pair is a screenshot of
         either the test, or one of its refs. We can identify which screenshot is
         for the test by comparing the url piece to the test name.

         Args:
            results_dir: str path to the directory that results live in
            test:name str name of the test that this artifact is for
            screenshot_artifact: list of strings, each being a url-png pair as
                described above.

         Returns:
              A dict mapping the screenshot key (ie: actual, expected) to the
              path of the file for that screenshot
         """
         result={}
         for screenshot_pair in screenshot_artifact:
             screenshot_split = screenshot_pair.split(":")
             url = screenshot_split[0]
             # The url produced by wptrunner will have a leading / which we trim
             # away for easier comparison to the test_name below.
             if url.startswith("/"):
                 url = url[1:]
             image_bytes = base64.b64decode(screenshot_split[1].strip())

             screenshot_key = "expected_image"
             file_suffix = test_failures.FILENAME_SUFFIX_EXPECTED
             if test_name == url:
                 screenshot_key = "actual_image"
                 file_suffix = test_failures.FILENAME_SUFFIX_ACTUAL

             screenshot_sub_path = (
                 os.path.join("layout-test-results",
                              self.port.output_filename(
                                  test_name, file_suffix, ".png"))
             )
             result[screenshot_key] = screenshot_sub_path

             screenshot_full_path = os.path.join(results_dir,screenshot_sub_path)
             if not os.path.exists(os.path.dirname(screenshot_full_path)):
                 os.makedirs(os.path.dirname(screenshot_full_path))
             # Note: we are writing raw bytes to this file
             with open(screenshot_full_path, "wb") as artifact_file:
                 artifact_file.write(image_bytes)
         return result
	# Copyright 2020 The Chromium Authors. All rights reserved.
	# Use of this source code is governed by a BSD-style license that can be
	# found in the LICENSE file.

	import base64
	import json
	import os
	import shutil
	import sys

	import common

	BLINK_TOOLS_DIR = os.path.join(common.SRC_DIR, 'third_party', 'blink', 'tools')
	WEB_TESTS_DIR = os.path.join(BLINK_TOOLS_DIR, os.pardir, 'web_tests')

	if BLINK_TOOLS_DIR not in sys.path:
	sys.path.append(BLINK_TOOLS_DIR)

	from blinkpy.common.host import Host
	from blinkpy.web_tests.models import test_failures

	class BaseWptScriptAdapter(common.BaseIsolatedScriptArgsAdapter):
	"""The base class for script adapters that use wptrunner to execute web
	platform tests. This contains any code shared between these scripts, such
	as integrating output with the results viewer. Subclasses contain other
	(usually platform-specific) logic."""

	def __init__(self):
	super(BaseWptScriptAdapter, self).__init__()
	host = Host()
	self.port = host.port_factory.get()

	def generate_test_output_args(self, output):
	return ['--log-chromium', output]

	def generate_sharding_args(self, total_shards, shard_index):
	return ['--total-chunks=%d' % total_shards,
	# shard_index is 0-based but WPT's this-chunk to be 1-based
	'--this-chunk=%d' % (shard_index + 1),
	# The default sharding strategy is to shard by directory. But
	# we want to hash each test to determine which shard runs it.
	# This allows running individual directories that have few
	# tests across many shards.
	'--chunk-type=hash']

	def do_post_test_run_tasks(self):
	# Move json results into layout-test-results directory
	results_dir = os.path.dirname(self.options.isolated_script_test_output)
	layout_test_results = os.path.join(results_dir, 'layout-test-results')
	if os.path.exists(layout_test_results):
	shutil.rmtree(layout_test_results)
	os.mkdir(layout_test_results)

	# Perform post-processing of wptrunner output
	self.process_wptrunner_output()

	shutil.copyfile(self.options.isolated_script_test_output,
	os.path.join(layout_test_results, 'full_results.json'))
	# create full_results_jsonp.js file which is used to
	# load results into the results viewer
	with open(self.options.isolated_script_test_output, 'r') \
	as full_results, \
	open(os.path.join(
	layout_test_results, 'full_results_jsonp.js'), 'w') \
	as json_js:
	json_js.write('ADD_FULL_RESULTS(%s);' % full_results.read())
	# copy layout test results viewer to layout-test-results directory
	shutil.copyfile(
	os.path.join(WEB_TESTS_DIR, 'fast', 'harness', 'results.html'),
	os.path.join(layout_test_results, 'results.html'))

	def process_wptrunner_output(self):
	"""Post-process the output generated by wptrunner.

	This output contains a single large json file containing the raw content
	or artifacts which need to be extracted into their own files and removed
	from the json file (to avoid duplication)."""
	output_json = json.load(
	open(self.options.isolated_script_test_output, "r"))
	test_json = output_json["tests"]
	results_dir = os.path.dirname(self.options.isolated_script_test_output)
	self._process_test_leaves(results_dir, output_json["path_delimiter"],
	test_json, "")
	# Write output_json back to the same file after modifying it in memory
	with open(self.options.isolated_script_test_output, "w") as output_file:
	json.dump(output_json, output_file)

	def _process_test_leaves(self, results_dir, delim, root_node, path_so_far):
	"""Finds and processes each test leaf below the specified root.

	This will recursively traverse the trie of results in the json output,
	keeping track of the path to each test and identifying leaves by the
	presence of certain attributes.

	Args:
	results_dir: str path to the dir that results are stored
	delim: str delimiter to be used for test names
	root_node: dict representing the root of the trie we're currently
	looking at
	path_so_far: str the path to the current root_node in the trie
	"""
	if "actual" in root_node:
	# Found a leaf, process it
	if "artifacts" not in root_node:
	return
	log_artifact = root_node["artifacts"].pop("log", None)
	if log_artifact:
	artifact_subpath = self._write_log_artifact(
	test_failures.FILENAME_SUFFIX_ACTUAL,
	results_dir, path_so_far, log_artifact)
	root_node["artifacts"]["actual_text"] = [artifact_subpath]

	screenshot_artifact = root_node["artifacts"].pop("screenshots",
	None)
	if screenshot_artifact:
	screenshot_paths_dict = self._write_screenshot_artifact(
	results_dir, path_so_far, screenshot_artifact)
	for screenshot_key, path in screenshot_paths_dict.items():
	root_node["artifacts"][screenshot_key] = [path]

	crashlog_artifact = root_node["artifacts"].pop("wpt_crash_log",
	None)
	if crashlog_artifact:
	artifact_subpath = self._write_log_artifact(
	test_failures.FILENAME_SUFFIX_CRASH_LOG,
	results_dir, path_so_far, crashlog_artifact)

	return

	# We're not at a leaf node, continue traversing the trie.
	for key in root_node:
	# Append the key to the path, separated by the delimiter. However if
	# the path is empty, skip the delimiter to avoid a leading slash in
	# the path.
	new_path = path_so_far + delim + key if path_so_far else key
	self._process_test_leaves(results_dir, delim, root_node[key],
	new_path)

	def _write_log_artifact(self, suffix, results_dir, test_name, log_artifact):
	"""Writes a log artifact to disk.

	A log artifact contains some form of output for a test. It is written to
	a txt file with a suffix generated from the log type.

	Args:
	suffix: str suffix of the artifact to write, e.g.
	test_failures.FILENAME_SUFFIX_ACTUAL
	results_dir: str path to the directory that results live in
	test_name: str name of the test that this artifact is for
	log_artifact: list of strings, the log entries for this test from
	the json output.

	Returns:
	string path to the artifact file that the log was written to,
	relative to the directory that the original output is located.
	"""
	log_artifact_sub_path = (
	os.path.join("layout-test-results",
	self.port.output_filename(test_name, suffix, ".txt"))
	)
	log_artifact_full_path = os.path.join(results_dir,
	log_artifact_sub_path)
	if not os.path.exists(os.path.dirname(log_artifact_full_path)):
	os.makedirs(os.path.dirname(log_artifact_full_path))
	with open(log_artifact_full_path, "w") as artifact_file:
	artifact_file.write("\n".join(log_artifact).encode("utf-8"))

	return log_artifact_sub_path

	def _write_screenshot_artifact(self, results_dir, test_name,
	screenshot_artifact):
	"""Write screenshot artifact to disk.

	The screenshot artifact is a list of strings, each of which has the
	format <url>:<base64-encoded PNG>. Each url-png pair is a screenshot of
	either the test, or one of its refs. We can identify which screenshot is
	for the test by comparing the url piece to the test name.

	Args:
	results_dir: str path to the directory that results live in
	test:name str name of the test that this artifact is for
	screenshot_artifact: list of strings, each being a url-png pair as
	described above.

	Returns:
	A dict mapping the screenshot key (ie: actual, expected) to the
	path of the file for that screenshot
	"""
	result={}
	for screenshot_pair in screenshot_artifact:
	screenshot_split = screenshot_pair.split(":")
	url = screenshot_split[0]
	# The url produced by wptrunner will have a leading / which we trim
	# away for easier comparison to the test_name below.
	if url.startswith("/"):
	url = url[1:]
	image_bytes = base64.b64decode(screenshot_split[1].strip())

	screenshot_key = "expected_image"
	file_suffix = test_failures.FILENAME_SUFFIX_EXPECTED
	if test_name == url:
	screenshot_key = "actual_image"
	file_suffix = test_failures.FILENAME_SUFFIX_ACTUAL

	screenshot_sub_path = (
	os.path.join("layout-test-results",
	self.port.output_filename(
	test_name, file_suffix, ".png"))
	)
	result[screenshot_key] = screenshot_sub_path

	screenshot_full_path = os.path.join(results_dir,screenshot_sub_path)
	if not os.path.exists(os.path.dirname(screenshot_full_path)):
	os.makedirs(os.path.dirname(screenshot_full_path))
	# Note: we are writing raw bytes to this file
	with open(screenshot_full_path, "wb") as artifact_file:
	artifact_file.write(image_bytes)
	return result