Tools/Scripts/webkit-test-results - WebKit - Git at Google

 #!/usr/bin/env python3
 #
 # Copyright (C) 2021 Igalia S.L.
 #
 # This program is free software; you can redistribute it and/or
 # modify it under the terms of the GNU Lesser General Public
 # License as published by the Free Software Foundation; either
 # version 2.1 of the License, or (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
 # Lesser General Public License for more details.
 #
 # You should have received a copy of the GNU Lesser General Public
 # License along with this program; if not, write to the
 # Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
 # Boston, MA 02110-1301, USA.
 # pylint: disable=missing-docstring,invalid-name

 """Command-line interface to the Webkit Results Database"""

 import argparse
 from datetime import datetime
 import json
 import logging
 import os
 import pathlib
 import sys
 import textwrap
 from urllib.parse import urljoin, urlencode, parse_qs
 from urllib.request import urlopen
 from urllib.error import HTTPError

 BASE_URL = "https://results.webkit.org/"


 def red(msg):
     """Returns a shell-aware red string"""
     return "\033[91m" + msg + "\033[0m"


 def green(msg):
     """Returns a shell-aware green string"""
     return "\033[92m" + msg + "\033[0m"


 # Roughly ordered from the most desired to the most unwanted.
 OUTCOMES = [
     "PASS",
     "TEXT",
     "IMAGE",
     "SKIP",
     "FAIL",
     "ERROR",
     "TIMEOUT",  # Put timeout as more severe than ERROR as they often are harder to debug
     "CRASH",
 ]


 def is_improvement(baseline, current):
     """Returns true if the current outcome is more favourable."""
     best_baseline = min(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
     return best_baseline > OUTCOMES.index(current)


 def is_regression(baseline, current):
     """Returns true if the current outcome downgraded the result."""
     worst_baseline = max(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
     return worst_baseline < OUTCOMES.index(current)


 def decorate(msg, baseline, current):
     """Colors the message if the status change represents an improvement or regression."""
     if is_improvement(baseline, current):
         msg = green(msg)
     elif is_regression(baseline, current):
         msg = red(msg)
     return msg


 class Query:
     """Wrapper around the most common query parameters for a given request"""

     def __init__(self, **kwargs):
         """Just stores the passed kwargs as parameters to be forwarded to the actual query"""
         self._params = kwargs.copy()

     @classmethod
     def from_query_string(cls, query_str):
         fields = parse_qs(query_str)
         # parse_qs forces the values to be lists, even when they are single-valued
         for k, v in fields.items():
             if len(v) == 1:
                 fields[k] = v[0]
         return cls(**fields)

     def as_query_string(self):
         """Escape and build the actual query string"""
         return urlencode(self._params)

     def add_param(self, **kwargs):
         """Extends the current set of parameters.

         Beware that it'll overwrite existing values."""
         self._params.update(kwargs)


 BOTS = {
     "gtk-release-x11": Query(
         platform="GTK", style="release", version_name="Xvfb", version="5.5.0"
     ),
     "gtk-release-gtk4": Query(
         platform="GTK", style="release", version_name="Xvfb", version="4.19.0"
     ),
     "gtk-release-wayland": Query(
         platform="GTK", style="release", version_name="Wayland"
     ),
     "wpe-release": Query(platform="WPE", style="release"),
     "wpe-debug": Query(platform="WPE", style="release"),
 }


 def get_commit_cache_filename():
     """Returns the cache filename, ensuring it's parent folder exists"""
     app_name = "webkit-test-results"
     xdg_config_home = os.getenv("XDG_CACHE_HOME")
     if xdg_config_home:
         directory = os.path.join(xdg_config_home, app_name)
     else:
         directory = os.path.join(os.path.expanduser("~"), ".cache", app_name)

     if not os.path.isdir(directory):
         pathlib.Path(directory).mkdir(parents=True, exist_ok=True)

     return os.path.join(directory, "commits.json")


 def load_commit_cache(force_reset=False):
     """Loads the stored commits, fetching again if needed."""
     filename = get_commit_cache_filename()
     if force_reset or not os.path.isfile(filename):
         return reset_commit_cache()

     logging.info("Opening commit cache %s", filename)
     with open(filename, encoding="utf-8") as handle:
         return json.load(handle)


 def reset_commit_cache():
     """Fetches the last 5000 commits and store their info to be reused in later calls."""
     # TODO Append to existing instead of resetting
     logging.info("Resetting commit cache")
     limit = 5000
     command = "/api/commits"

     query = {"limit": limit, "branch": "main"}

     url = urljoin(BASE_URL, command) + "?" + urlencode(query)

     filename = get_commit_cache_filename()
     logging.info("Fetching commits from %s", url)
     with urlopen(url) as response:
         raw_data = json.load(response)

     uuids = {}
     for commit in raw_data:
         # Order is usually zero for single commits pushed to the repo. For example,
         # commits in SVN. When moving to git, branchs with multiple commits will
         # make use of it.
         uuid = str(commit["timestamp"] * 100 + commit["order"])
         commit["uuid"] = uuid
         uuids[uuid] = commit

     logging.info("Saving commits to %s", filename)
     with open(filename, "w", encoding="utf-8") as output:
         logging.info("Saving downloaded cache")
         json.dump(uuids, output)

     return uuids


 def last_run(args):
     """Shows data about the last test run registered for the selected bot"""
     endpoint = "api/results/layout-tests"
     if args.only_changes:
         logging.info("--only-changes ignored in this command")
     if args.only_unexpected:
         logging.info("--only-unexpected ignored in this command")

     configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
     configuration.add_param(limit=1)
     query = configuration.as_query_string()

     url = urljoin(BASE_URL, endpoint) + "?" + query

     logging.info("Loading test data from %s", url)
     try:
         with urlopen(url) as response:
             data = json.load(response)
         print(json.dumps(data, sort_keys=True, indent=4, separators=(",", ": ")))
     except (HTTPError, json.JSONDecodeError) as e:
         print(e)
         return 1

     return 0


 def get_latest_commit(commits):
     """Returns the newest commit from a set of commits"""

     latest = max(commits.values(), key=lambda x: x["uuid"])

     return latest["identifier"], latest["timestamp"]


 def report_test(args):
     """Reports the test history for a single testcase in a single configuration"""
     endpoint = "api/results/layout-tests"
     configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
     if args.limit > 0:
         configuration.add_param(limit=args.limit)
     query = configuration.as_query_string()

     url = urljoin(BASE_URL, endpoint + "/" + args.test) + "?" + query

     logging.info("Loading test data from %s", url)
     try:
         with urlopen(url) as response:
             data = json.load(response)[0]
     except IndexError:
         logging.error("No results returned. Exiting.")
         return 1

     results = sorted(data["results"], key=lambda result: result["uuid"])

     commits = load_commit_cache(force_reset=args.reset_cache)
     logging.info("Found %d cached commits", len(commits))

     previous = None

     matched = False

     for result in results:
         actual = result["actual"]
         expected = result["expected"]
         start_time = datetime.fromtimestamp(result["start_time"])
         uuid = result["uuid"]

         try:
             commit = commits[str(uuid)]
             matched = True
         except KeyError:
             logging.info("Could not find commit with uuid %s", uuid)
             continue

         if args.only_changes:
             if previous == actual:
                 continue
         elif args.only_unexpected:
             if actual == expected:
                 continue

         msg = "commit {} expected {} actual {} time {}".format(
             commit["identifier"], expected, actual, start_time
         )

         if args.color:
             print(decorate(msg, expected, actual))
         else:
             print(msg)
         previous = actual

     if not matched:
         latest_revision, latest_timestamp = get_latest_commit(commits)
         latest_date = datetime.fromtimestamp(latest_timestamp)
         print(
             f"No commits matched. The latest commit in cache is {latest_revision} from {latest_date}",
             file=sys.stderr,
         )
         return 1

     return 0


 def parse_args():
     """Parse command line arguments and switches"""
     parser = argparse.ArgumentParser(
         description=textwrap.dedent(
             """
                 Command line tool to query https://results.webkit.org for test history.

                 Passing a test case in the command line will show one entry for each
                 test run registered, alongside commit information, expected and actual outcomes,
                 and timestamp of the run. For example:

                 $ wk-test-result --bot wpe-release ietestcenter/css3/text/textshadow-001.htm --limit=5
                 commit 244781@main expected PASS actual TEXT time 2021-12-02 18:02:13
                 commit 244787@main expected PASS actual TEXT time 2021-12-02 19:43:10
                 commit 244796@main expected PASS actual TEXT time 2021-12-02 20:41:52
                 commit 244797@main expected PASS actual TEXT time 2021-12-02 21:40:29
                 commit 244803@main expected PASS actual TEXT time 2021-12-02 23:13:19

                 To avoid querying the commit data at each invocation, the last 5000 commits
                 are cached (by default, to $XDG_CACHE_HOME/wk-gardening-tools/commits.json).

                 Currently, only GLIB-based bots and layout-test suite are supported.
             """
         ),
         formatter_class=argparse.RawTextHelpFormatter,
     )

     bots_group = parser.add_mutually_exclusive_group()
     bots_group.add_argument("-b", "--bot", help="Bot to query")
     bots_group.add_argument("--list-bots", action="store_true", help="List predefined bots")

     # Common options
     parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
     parser.add_argument(
         "-r",
         "--reset-cache",
         action="store_true",
         help="Reset the commit cache if needed",
     )
     parser.add_argument(
         "-c",
         "--color",
         action="store_true",
         help="Use colors to highlight unexpected results",
     )

     parser.add_argument(
         "--last-run",
         action="store_true",
         help="Show status of the last registered run and exit.",
     )

     parser.add_argument(
         "--limit",
         type=int,
         default=100,
         help="""Limit the number of results. Defaults to 100, use -1 to ask the server for
                 its default value. Use together with --test""",
     )


     # Output options
     output_group = parser.add_mutually_exclusive_group()
     output_group.add_argument(
         "--only-changes",
         action="store_true",
         help="Display only revisions where the state changed. Use with --test",
     )
     output_group.add_argument(
         "--only-unexpected",
         action="store_true",
         help="Display only unexpected results. Use with --test",
     )

     parser.add_argument(
         "test", help="Test case to be searched", nargs="?", default=None
     )

     return parser.parse_args()


 def main():
     """Main entry point"""
     args = parse_args()

     if args.verbose:
         logging.basicConfig(level=logging.DEBUG)

     if args.list_bots:
         for bot in sorted(BOTS.keys()):
             print(bot)
     elif args.last_run:
         sys.exit(last_run(args))
     elif args.test:
         if not args.bot:
             print("Test history requires a bot to be specified with -b or --bot. Exiting.")
             sys.exit(1)
         sys.exit(report_test(args))
     else:
         print("No test case provided. Did you forget to pass it?")
         sys.exit(1)


 if __name__ == "__main__":
     main()
	#!/usr/bin/env python3
	#
	# Copyright (C) 2021 Igalia S.L.
	#
	# This program is free software; you can redistribute it and/or
	# modify it under the terms of the GNU Lesser General Public
	# License as published by the Free Software Foundation; either
	# version 2.1 of the License, or (at your option) any later version.
	#
	# This program is distributed in the hope that it will be useful,
	# but WITHOUT ANY WARRANTY; without even the implied warranty of
	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
	# Lesser General Public License for more details.
	#
	# You should have received a copy of the GNU Lesser General Public
	# License along with this program; if not, write to the
	# Free Software Foundation, Inc., 51 Franklin St, Fifth Floor,
	# Boston, MA 02110-1301, USA.
	# pylint: disable=missing-docstring,invalid-name

	"""Command-line interface to the Webkit Results Database"""

	import argparse
	from datetime import datetime
	import json
	import logging
	import os
	import pathlib
	import sys
	import textwrap
	from urllib.parse import urljoin, urlencode, parse_qs
	from urllib.request import urlopen
	from urllib.error import HTTPError

	BASE_URL = "https://results.webkit.org/"


	def red(msg):
	"""Returns a shell-aware red string"""
	return "\033[91m" + msg + "\033[0m"


	def green(msg):
	"""Returns a shell-aware green string"""
	return "\033[92m" + msg + "\033[0m"


	# Roughly ordered from the most desired to the most unwanted.
	OUTCOMES = [
	"PASS",
	"TEXT",
	"IMAGE",
	"SKIP",
	"FAIL",
	"ERROR",
	"TIMEOUT", # Put timeout as more severe than ERROR as they often are harder to debug
	"CRASH",
	]


	def is_improvement(baseline, current):
	"""Returns true if the current outcome is more favourable."""
	best_baseline = min(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
	return best_baseline > OUTCOMES.index(current)


	def is_regression(baseline, current):
	"""Returns true if the current outcome downgraded the result."""
	worst_baseline = max(OUTCOMES.index(baseline_token) for baseline_token in baseline.split())
	return worst_baseline < OUTCOMES.index(current)


	def decorate(msg, baseline, current):
	"""Colors the message if the status change represents an improvement or regression."""
	if is_improvement(baseline, current):
	msg = green(msg)
	elif is_regression(baseline, current):
	msg = red(msg)
	return msg


	class Query:
	"""Wrapper around the most common query parameters for a given request"""

	def __init__(self, **kwargs):
	"""Just stores the passed kwargs as parameters to be forwarded to the actual query"""
	self._params = kwargs.copy()

	@classmethod
	def from_query_string(cls, query_str):
	fields = parse_qs(query_str)
	# parse_qs forces the values to be lists, even when they are single-valued
	for k, v in fields.items():
	if len(v) == 1:
	fields[k] = v[0]
	return cls(**fields)

	def as_query_string(self):
	"""Escape and build the actual query string"""
	return urlencode(self._params)

	def add_param(self, **kwargs):
	"""Extends the current set of parameters.

	Beware that it'll overwrite existing values."""
	self._params.update(kwargs)


	BOTS = {
	"gtk-release-x11": Query(
	platform="GTK", style="release", version_name="Xvfb", version="5.5.0"
	),
	"gtk-release-gtk4": Query(
	platform="GTK", style="release", version_name="Xvfb", version="4.19.0"
	),
	"gtk-release-wayland": Query(
	platform="GTK", style="release", version_name="Wayland"
	),
	"wpe-release": Query(platform="WPE", style="release"),
	"wpe-debug": Query(platform="WPE", style="release"),
	}


	def get_commit_cache_filename():
	"""Returns the cache filename, ensuring it's parent folder exists"""
	app_name = "webkit-test-results"
	xdg_config_home = os.getenv("XDG_CACHE_HOME")
	if xdg_config_home:
	directory = os.path.join(xdg_config_home, app_name)
	else:
	directory = os.path.join(os.path.expanduser("~"), ".cache", app_name)

	if not os.path.isdir(directory):
	pathlib.Path(directory).mkdir(parents=True, exist_ok=True)

	return os.path.join(directory, "commits.json")


	def load_commit_cache(force_reset=False):
	"""Loads the stored commits, fetching again if needed."""
	filename = get_commit_cache_filename()
	if force_reset or not os.path.isfile(filename):
	return reset_commit_cache()

	logging.info("Opening commit cache %s", filename)
	with open(filename, encoding="utf-8") as handle:
	return json.load(handle)


	def reset_commit_cache():
	"""Fetches the last 5000 commits and store their info to be reused in later calls."""
	# TODO Append to existing instead of resetting
	logging.info("Resetting commit cache")
	limit = 5000
	command = "/api/commits"

	query = {"limit": limit, "branch": "main"}

	url = urljoin(BASE_URL, command) + "?" + urlencode(query)

	filename = get_commit_cache_filename()
	logging.info("Fetching commits from %s", url)
	with urlopen(url) as response:
	raw_data = json.load(response)

	uuids = {}
	for commit in raw_data:
	# Order is usually zero for single commits pushed to the repo. For example,
	# commits in SVN. When moving to git, branchs with multiple commits will
	# make use of it.
	uuid = str(commit["timestamp"] * 100 + commit["order"])
	commit["uuid"] = uuid
	uuids[uuid] = commit

	logging.info("Saving commits to %s", filename)
	with open(filename, "w", encoding="utf-8") as output:
	logging.info("Saving downloaded cache")
	json.dump(uuids, output)

	return uuids


	def last_run(args):
	"""Shows data about the last test run registered for the selected bot"""
	endpoint = "api/results/layout-tests"
	if args.only_changes:
	logging.info("--only-changes ignored in this command")
	if args.only_unexpected:
	logging.info("--only-unexpected ignored in this command")

	configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
	configuration.add_param(limit=1)
	query = configuration.as_query_string()

	url = urljoin(BASE_URL, endpoint) + "?" + query

	logging.info("Loading test data from %s", url)
	try:
	with urlopen(url) as response:
	data = json.load(response)
	print(json.dumps(data, sort_keys=True, indent=4, separators=(",", ": ")))
	except (HTTPError, json.JSONDecodeError) as e:
	print(e)
	return 1

	return 0


	def get_latest_commit(commits):
	"""Returns the newest commit from a set of commits"""

	latest = max(commits.values(), key=lambda x: x["uuid"])

	return latest["identifier"], latest["timestamp"]


	def report_test(args):
	"""Reports the test history for a single testcase in a single configuration"""
	endpoint = "api/results/layout-tests"
	configuration = BOTS.get(args.bot, Query.from_query_string(args.bot))
	if args.limit > 0:
	configuration.add_param(limit=args.limit)
	query = configuration.as_query_string()

	url = urljoin(BASE_URL, endpoint + "/" + args.test) + "?" + query

	logging.info("Loading test data from %s", url)
	try:
	with urlopen(url) as response:
	data = json.load(response)[0]
	except IndexError:
	logging.error("No results returned. Exiting.")
	return 1

	results = sorted(data["results"], key=lambda result: result["uuid"])

	commits = load_commit_cache(force_reset=args.reset_cache)
	logging.info("Found %d cached commits", len(commits))

	previous = None

	matched = False

	for result in results:
	actual = result["actual"]
	expected = result["expected"]
	start_time = datetime.fromtimestamp(result["start_time"])
	uuid = result["uuid"]

	try:
	commit = commits[str(uuid)]
	matched = True
	except KeyError:
	logging.info("Could not find commit with uuid %s", uuid)
	continue

	if args.only_changes:
	if previous == actual:
	continue
	elif args.only_unexpected:
	if actual == expected:
	continue

	msg = "commit {} expected {} actual {} time {}".format(
	commit["identifier"], expected, actual, start_time
	)

	if args.color:
	print(decorate(msg, expected, actual))
	else:
	print(msg)
	previous = actual

	if not matched:
	latest_revision, latest_timestamp = get_latest_commit(commits)
	latest_date = datetime.fromtimestamp(latest_timestamp)
	print(
	f"No commits matched. The latest commit in cache is {latest_revision} from {latest_date}",
	file=sys.stderr,
	)
	return 1

	return 0


	def parse_args():
	"""Parse command line arguments and switches"""
	parser = argparse.ArgumentParser(
	description=textwrap.dedent(
	"""
	Command line tool to query https://results.webkit.org for test history.

	Passing a test case in the command line will show one entry for each
	test run registered, alongside commit information, expected and actual outcomes,
	and timestamp of the run. For example:

	$ wk-test-result --bot wpe-release ietestcenter/css3/text/textshadow-001.htm --limit=5
	commit 244781@main expected PASS actual TEXT time 2021-12-02 18:02:13
	commit 244787@main expected PASS actual TEXT time 2021-12-02 19:43:10
	commit 244796@main expected PASS actual TEXT time 2021-12-02 20:41:52
	commit 244797@main expected PASS actual TEXT time 2021-12-02 21:40:29
	commit 244803@main expected PASS actual TEXT time 2021-12-02 23:13:19

	To avoid querying the commit data at each invocation, the last 5000 commits
	are cached (by default, to $XDG_CACHE_HOME/wk-gardening-tools/commits.json).

	Currently, only GLIB-based bots and layout-test suite are supported.
	"""
	),
	formatter_class=argparse.RawTextHelpFormatter,
	)

	bots_group = parser.add_mutually_exclusive_group()
	bots_group.add_argument("-b", "--bot", help="Bot to query")
	bots_group.add_argument("--list-bots", action="store_true", help="List predefined bots")

	# Common options
	parser.add_argument("-v", "--verbose", action="store_true", help="Verbose output")
	parser.add_argument(
	"-r",
	"--reset-cache",
	action="store_true",
	help="Reset the commit cache if needed",
	)
	parser.add_argument(
	"-c",
	"--color",
	action="store_true",
	help="Use colors to highlight unexpected results",
	)

	parser.add_argument(
	"--last-run",
	action="store_true",
	help="Show status of the last registered run and exit.",
	)

	parser.add_argument(
	"--limit",
	type=int,
	default=100,
	help="""Limit the number of results. Defaults to 100, use -1 to ask the server for
	its default value. Use together with --test""",
	)


	# Output options
	output_group = parser.add_mutually_exclusive_group()
	output_group.add_argument(
	"--only-changes",
	action="store_true",
	help="Display only revisions where the state changed. Use with --test",
	)
	output_group.add_argument(
	"--only-unexpected",
	action="store_true",
	help="Display only unexpected results. Use with --test",
	)

	parser.add_argument(
	"test", help="Test case to be searched", nargs="?", default=None
	)

	return parser.parse_args()


	def main():
	"""Main entry point"""
	args = parse_args()

	if args.verbose:
	logging.basicConfig(level=logging.DEBUG)

	if args.list_bots:
	for bot in sorted(BOTS.keys()):
	print(bot)
	elif args.last_run:
	sys.exit(last_run(args))
	elif args.test:
	if not args.bot:
	print("Test history requires a bot to be specified with -b or --bot. Exiting.")
	sys.exit(1)
	sys.exit(report_test(args))
	else:
	print("No test case provided. Did you forget to pass it?")
	sys.exit(1)


	if __name__ == "__main__":
	main()