static_analysis/checkpatch_presubmit_bazel.py - kernel/build - Git at Google

 # Copyright (C) 2023 The Android Open Source Project
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #       http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # distributed under the License is distributed on an "AS IS" BASIS,
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.

 """Runs necessary checkpatch targets for a build on ci.android.com.

 usage:

 tools/bazel run //build/kernel/static_analysis:checkpatch_presubmit -- \\
     --dist_dir <DIST_DIR> \\
     [<other flags to checkpatch>] \\
 """

 import argparse
 import collections
 import json
 import logging
 import os
 import pathlib
 import shlex
 import subprocess
 import sys
 from typing import Any

 _LOG_LEVEL = logging.INFO
 # _LOG_LEVEL = logging.DEBUG

 _BAZEL = pathlib.Path("tools/bazel")
 _SILENT_ARGS = [
     "--ui_event_filters=-info",
     "--noshow_progress",
 ]

 # TODO: Find a better way to handle this exceptions;
 _PATH_PREFIX_DENY_LIST = (
     "external/",
     "bootable/",
     "prebuilts/fuchsia_sdk",
 )

 def load_arguments() -> dict[str, Any]:
     parser = argparse.ArgumentParser(
         description=__doc__,
         formatter_class=argparse.RawTextHelpFormatter)
     parser.add_argument(
         "--dist_dir",
         type=_resolve_against_workspace_root,
         required=True,
         help="DIST_DIR. If relative, resolve against workspace root.",
     )
     parser.add_argument(
         "--bid",
         help="Build ID. If specified, it is used to skip the check on post-submit.",
     )
     parser.add_argument(
         "--change_info",
         type=_require_absolute_path,
         help="Path to change-info file providing complete change information.",
     )
     return parser.parse_known_args()


 def _resolve_against_workspace_root(value: str) -> pathlib.Path:
     path = pathlib.Path(value)
     if path.is_absolute():
         return path
     return pathlib.Path(os.environ["BUILD_WORKSPACE_DIRECTORY"]) / path

 def _require_absolute_path(p: str | pathlib.Path) -> pathlib.Path:
     p = pathlib.Path(p)
     if not p.is_absolute():
         raise argparse.ArgumentTypeError("need to specify an absolute path")
     return p

 def _log_command(args):
     quoted = [shlex.quote(str(arg)) for arg in args]
     logging.debug("Running command line: %s", " ".join(quoted))


 def _find_checkpatch_targets(path: pathlib.Path) -> list[str]:
     if str(path).startswith(_PATH_PREFIX_DENY_LIST):
         logging.info("Skipped //%s path in deny list", path)
         return []

     if not _resolve_against_workspace_root(path / "BUILD.bazel").is_file() and \
         not _resolve_against_workspace_root(path / "BUILD").is_file():
         logging.info("//%s is not a package; no BUILD file is found", path)
         return []

     args = [_BAZEL, "query"]
     args += _SILENT_ARGS
     args.append(f'kind("^checkpatch rule$", //{path}:all)')
     _log_command(args)
     lines = subprocess.check_output(
         args,
         text=True,
         cwd=_resolve_against_workspace_root("."),
     ).splitlines()
     return [line.strip() for line in lines if line.strip()]


 def _run_checkpatch(
     target: str,
     git_sha1: str,
     log: pathlib.Path,
     checkpatch_args: list[str],
     silent: bool = False,
 ) -> int:
     args = [_BAZEL, "run", "--show_result=0"]
     args += _SILENT_ARGS
     args += [target, "--"]
     args += checkpatch_args
     args += ["--log", log]
     args += ["--git_sha1", git_sha1]
     _log_command(args)
     return subprocess.run(
         args,
         text=True,
         cwd=_resolve_against_workspace_root("."),
         stdout=subprocess.DEVNULL if silent else None,
         stderr=subprocess.STDOUT if silent else None,
     ).returncode


 def _find_repo(curdir: pathlib.Path) -> pathlib.Path | None:
     """Find repo installation."""
     while curdir.parent != curdir:  # is not root
         maybe_dot_repo = curdir / ".repo"
         if maybe_dot_repo.is_dir():
             return curdir
         curdir = curdir.parent
     return None


 def _get_package_path(path: pathlib.Path) -> pathlib.Path | None:
     """Get package path from project path"""
     workspace_dir = pathlib.Path(os.environ["BUILD_WORKSPACE_DIRECTORY"])
     repo_root_s = os.environ.get("KLEAF_REPO_MANIFEST", ":").split(":")[0]
     if repo_root_s:
         repo_root = pathlib.Path(repo_root_s).resolve()
     else:
         repo_root = _find_repo(workspace_dir)

     if not repo_root:
         logging.error(
             "Unable to determine repo root. Please specify --repo_manifest.")
         return None

     realpath = repo_root / path
     if realpath.is_relative_to(workspace_dir):
         return realpath.relative_to(workspace_dir)

     return None


 def _invoke_using_applied_prop(dist_dir: pathlib.Path) -> int:
     applied_prop = dist_dir / "applied.prop"
     applied_prop_dict: dict[pathlib.Path, list[str]] = \
         collections.defaultdict(list)
     with open(applied_prop) as applied_prop_file:
         for line in applied_prop_file:
             line = line.strip()
             if not line:
                 continue
             path, git_sha1 = line.split(maxsplit=2)
             applied_prop_dict[pathlib.Path(path)].append(git_sha1)

     targets: list[(list[str], str)] = []
     for path, git_sha1_list in applied_prop_dict.items():
         if len(git_sha1_list) > 1:
             logging.error("Multiple git sha1 found in %s for %s",
                           applied_prop, path)
             return 1

         package_path = _get_package_path(path)
         if not package_path:
             logging.info("Skipping %s because it is not in the workspace.", path)
             continue

         path_targets = _find_checkpatch_targets(package_path)
         if not path_targets:
             logging.info(
                 "Skipping %s because no checkpatch() target is found.", path)
             continue
         targets.append((path_targets, git_sha1_list[0]))

     checkpatch_log = dist_dir / "checkpatch.log"
     checkpatch_full_log = dist_dir / "checkpatch_full.log"
     if checkpatch_log.exists():
         os.unlink(checkpatch_log)
     if checkpatch_full_log.exists():
         os.unlink(checkpatch_full_log)
     return_codes = []
     for path_targets, git_sha1 in targets:
         for target in path_targets:
             return_codes.append(_run_checkpatch(
                 target=target,
                 git_sha1=git_sha1,
                 log=checkpatch_log,
                 checkpatch_args=checkpatch_args,
             ))
             _run_checkpatch(
                 target=target,
                 git_sha1=git_sha1,
                 log=checkpatch_full_log,
                 checkpatch_args=checkpatch_args + ["--ignored_checks", ""],
                 silent=True,
             )

     success = sum(return_codes) == 0

     if not success:
         logging.info("See %s for complete output.", checkpatch_log.name)

     return success


 def _invoke_using_change_info_json(
     dist_dir: pathlib.Path,
     change_info: pathlib.Path
 ) -> int:
     targets: list[(list[str], str, str)] = []
     with change_info.open() as change_info_file:
         for change in json.load(change_info_file).get("changes"):
             project_name = change["project"]
             project_path = pathlib.Path(change["projectPath"])

             package_path = _get_package_path(project_path)
             if not package_path:
                 logging.info("Skipping %s because it is not in the workspace.", project_path)
                 continue

             # Only interested in the git SHA of the CL at the time of the
             # build. The SHA is specified by the "latestRevision" field.
             revision = change["latestRevision"]

             path_targets = _find_checkpatch_targets(package_path)
             if not path_targets:
                 logging.info(
                     "Skipping %s because no checkpatch() target is found.", project_path)
                 continue

             targets.append((path_targets, revision, project_name))

     checkpatch_topdir = dist_dir / "checkpatch"
     return_codes = []

     for path_targets, git_sha1, project_name in targets:
         sanitized_project_name = project_name.replace("/", "__")
         checkpatch_dir = checkpatch_topdir / sanitized_project_name / git_sha1
         checkpatch_dir.mkdir(parents=True, exist_ok=True)

         checkpatch_log = checkpatch_dir / "checkpatch.log"
         checkpatch_log.unlink(missing_ok=True)

         checkpatch_full_log = checkpatch_dir / "checkpatch_full.log"
         checkpatch_full_log.unlink(missing_ok=True)

         for target in path_targets:
             return_codes.append(_run_checkpatch(
                 target=target,
                 git_sha1=git_sha1,
                 log=checkpatch_log,
                 checkpatch_args=checkpatch_args,
             ))
             _run_checkpatch(
                 target=target,
                 git_sha1=git_sha1,
                 log=checkpatch_full_log,
                 checkpatch_args=checkpatch_args + ["--ignored_checks", ""],
                 silent=True,
             )

     success = sum(return_codes) == 0

     if not success:
         logging.info(
             "See %s/ folder for complete output.",
             checkpatch_topdir.name
         )

     return success

 def main(
         checkpatch_args: list[str],
         dist_dir: pathlib.Path,
         bid: str | None,
         change_info: pathlib.Path | None,
 ) -> int:
     if bid:
         # Skip checkpatch for postsubmit (b/35390488).
         if not bid.startswith("P"):
             logging.info("Did not identify a presubmit build. Exiting.")
             return 0

     if change_info:
         success = _invoke_using_change_info_json(
             dist_dir=dist_dir,
             change_info=change_info
         )
     else:
         # Fallback, since change-info is not guaranteed to exist
         success = _invoke_using_applied_prop(dist_dir=dist_dir)

     return 0 if success else 1


 if __name__ == "__main__":
     logging.basicConfig(level=_LOG_LEVEL,
                         format="%(levelname)s: %(message)s")
     known, checkpatch_args = load_arguments()
     sys.exit(main(checkpatch_args=checkpatch_args, **vars(known)))
	# Copyright (C) 2023 The Android Open Source Project
	#
	# Licensed under the Apache License, Version 2.0 (the "License");
	# you may not use this file except in compliance with the License.
	# You may obtain a copy of the License at
	#
	# http://www.apache.org/licenses/LICENSE-2.0
	#
	# Unless required by applicable law or agreed to in writing, software
	# distributed under the License is distributed on an "AS IS" BASIS,
	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	# See the License for the specific language governing permissions and
	# limitations under the License.

	"""Runs necessary checkpatch targets for a build on ci.android.com.

	usage:

	tools/bazel run //build/kernel/static_analysis:checkpatch_presubmit -- \\
	--dist_dir <DIST_DIR> \\
	[<other flags to checkpatch>] \\
	"""

	import argparse
	import collections
	import json
	import logging
	import os
	import pathlib
	import shlex
	import subprocess
	import sys
	from typing import Any

	_LOG_LEVEL = logging.INFO
	# _LOG_LEVEL = logging.DEBUG

	_BAZEL = pathlib.Path("tools/bazel")
	_SILENT_ARGS = [
	"--ui_event_filters=-info",
	"--noshow_progress",
	]

	# TODO: Find a better way to handle this exceptions;
	_PATH_PREFIX_DENY_LIST = (
	"external/",
	"bootable/",
	"prebuilts/fuchsia_sdk",
	)

	def load_arguments() -> dict[str, Any]:
	parser = argparse.ArgumentParser(
	description=__doc__,
	formatter_class=argparse.RawTextHelpFormatter)
	parser.add_argument(
	"--dist_dir",
	type=_resolve_against_workspace_root,
	required=True,
	help="DIST_DIR. If relative, resolve against workspace root.",
	)
	parser.add_argument(
	"--bid",
	help="Build ID. If specified, it is used to skip the check on post-submit.",
	)
	parser.add_argument(
	"--change_info",
	type=_require_absolute_path,
	help="Path to change-info file providing complete change information.",
	)
	return parser.parse_known_args()


	def _resolve_against_workspace_root(value: str) -> pathlib.Path:
	path = pathlib.Path(value)
	if path.is_absolute():
	return path
	return pathlib.Path(os.environ["BUILD_WORKSPACE_DIRECTORY"]) / path

	def _require_absolute_path(p: str \| pathlib.Path) -> pathlib.Path:
	p = pathlib.Path(p)
	if not p.is_absolute():
	raise argparse.ArgumentTypeError("need to specify an absolute path")
	return p

	def _log_command(args):
	quoted = [shlex.quote(str(arg)) for arg in args]
	logging.debug("Running command line: %s", " ".join(quoted))


	def _find_checkpatch_targets(path: pathlib.Path) -> list[str]:
	if str(path).startswith(_PATH_PREFIX_DENY_LIST):
	logging.info("Skipped //%s path in deny list", path)
	return []

	if not _resolve_against_workspace_root(path / "BUILD.bazel").is_file() and \
	not _resolve_against_workspace_root(path / "BUILD").is_file():
	logging.info("//%s is not a package; no BUILD file is found", path)
	return []

	args = [_BAZEL, "query"]
	args += _SILENT_ARGS
	args.append(f'kind("^checkpatch rule$", //{path}:all)')
	_log_command(args)
	lines = subprocess.check_output(
	args,
	text=True,
	cwd=_resolve_against_workspace_root("."),
	).splitlines()
	return [line.strip() for line in lines if line.strip()]


	def _run_checkpatch(
	target: str,
	git_sha1: str,
	log: pathlib.Path,
	checkpatch_args: list[str],
	silent: bool = False,
	) -> int:
	args = [_BAZEL, "run", "--show_result=0"]
	args += _SILENT_ARGS
	args += [target, "--"]
	args += checkpatch_args
	args += ["--log", log]
	args += ["--git_sha1", git_sha1]
	_log_command(args)
	return subprocess.run(
	args,
	text=True,
	cwd=_resolve_against_workspace_root("."),
	stdout=subprocess.DEVNULL if silent else None,
	stderr=subprocess.STDOUT if silent else None,
	).returncode


	def _find_repo(curdir: pathlib.Path) -> pathlib.Path \| None:
	"""Find repo installation."""
	while curdir.parent != curdir: # is not root
	maybe_dot_repo = curdir / ".repo"
	if maybe_dot_repo.is_dir():
	return curdir
	curdir = curdir.parent
	return None


	def _get_package_path(path: pathlib.Path) -> pathlib.Path \| None:
	"""Get package path from project path"""
	workspace_dir = pathlib.Path(os.environ["BUILD_WORKSPACE_DIRECTORY"])
	repo_root_s = os.environ.get("KLEAF_REPO_MANIFEST", ":").split(":")[0]
	if repo_root_s:
	repo_root = pathlib.Path(repo_root_s).resolve()
	else:
	repo_root = _find_repo(workspace_dir)

	if not repo_root:
	logging.error(
	"Unable to determine repo root. Please specify --repo_manifest.")
	return None

	realpath = repo_root / path
	if realpath.is_relative_to(workspace_dir):
	return realpath.relative_to(workspace_dir)

	return None


	def _invoke_using_applied_prop(dist_dir: pathlib.Path) -> int:
	applied_prop = dist_dir / "applied.prop"
	applied_prop_dict: dict[pathlib.Path, list[str]] = \
	collections.defaultdict(list)
	with open(applied_prop) as applied_prop_file:
	for line in applied_prop_file:
	line = line.strip()
	if not line:
	continue
	path, git_sha1 = line.split(maxsplit=2)
	applied_prop_dict[pathlib.Path(path)].append(git_sha1)

	targets: list[(list[str], str)] = []
	for path, git_sha1_list in applied_prop_dict.items():
	if len(git_sha1_list) > 1:
	logging.error("Multiple git sha1 found in %s for %s",
	applied_prop, path)
	return 1

	package_path = _get_package_path(path)
	if not package_path:
	logging.info("Skipping %s because it is not in the workspace.", path)
	continue

	path_targets = _find_checkpatch_targets(package_path)
	if not path_targets:
	logging.info(
	"Skipping %s because no checkpatch() target is found.", path)
	continue
	targets.append((path_targets, git_sha1_list[0]))

	checkpatch_log = dist_dir / "checkpatch.log"
	checkpatch_full_log = dist_dir / "checkpatch_full.log"
	if checkpatch_log.exists():
	os.unlink(checkpatch_log)
	if checkpatch_full_log.exists():
	os.unlink(checkpatch_full_log)
	return_codes = []
	for path_targets, git_sha1 in targets:
	for target in path_targets:
	return_codes.append(_run_checkpatch(
	target=target,
	git_sha1=git_sha1,
	log=checkpatch_log,
	checkpatch_args=checkpatch_args,
	))
	_run_checkpatch(
	target=target,
	git_sha1=git_sha1,
	log=checkpatch_full_log,
	checkpatch_args=checkpatch_args + ["--ignored_checks", ""],
	silent=True,
	)

	success = sum(return_codes) == 0

	if not success:
	logging.info("See %s for complete output.", checkpatch_log.name)

	return success


	def _invoke_using_change_info_json(
	dist_dir: pathlib.Path,
	change_info: pathlib.Path
	) -> int:
	targets: list[(list[str], str, str)] = []
	with change_info.open() as change_info_file:
	for change in json.load(change_info_file).get("changes"):
	project_name = change["project"]
	project_path = pathlib.Path(change["projectPath"])

	package_path = _get_package_path(project_path)
	if not package_path:
	logging.info("Skipping %s because it is not in the workspace.", project_path)
	continue

	# Only interested in the git SHA of the CL at the time of the
	# build. The SHA is specified by the "latestRevision" field.
	revision = change["latestRevision"]

	path_targets = _find_checkpatch_targets(package_path)
	if not path_targets:
	logging.info(
	"Skipping %s because no checkpatch() target is found.", project_path)
	continue

	targets.append((path_targets, revision, project_name))

	checkpatch_topdir = dist_dir / "checkpatch"
	return_codes = []

	for path_targets, git_sha1, project_name in targets:
	sanitized_project_name = project_name.replace("/", "__")
	checkpatch_dir = checkpatch_topdir / sanitized_project_name / git_sha1
	checkpatch_dir.mkdir(parents=True, exist_ok=True)

	checkpatch_log = checkpatch_dir / "checkpatch.log"
	checkpatch_log.unlink(missing_ok=True)

	checkpatch_full_log = checkpatch_dir / "checkpatch_full.log"
	checkpatch_full_log.unlink(missing_ok=True)

	for target in path_targets:
	return_codes.append(_run_checkpatch(
	target=target,
	git_sha1=git_sha1,
	log=checkpatch_log,
	checkpatch_args=checkpatch_args,
	))
	_run_checkpatch(
	target=target,
	git_sha1=git_sha1,
	log=checkpatch_full_log,
	checkpatch_args=checkpatch_args + ["--ignored_checks", ""],
	silent=True,
	)

	success = sum(return_codes) == 0

	if not success:
	logging.info(
	"See %s/ folder for complete output.",
	checkpatch_topdir.name
	)

	return success

	def main(
	checkpatch_args: list[str],
	dist_dir: pathlib.Path,
	bid: str \| None,
	change_info: pathlib.Path \| None,
	) -> int:
	if bid:
	# Skip checkpatch for postsubmit (b/35390488).
	if not bid.startswith("P"):
	logging.info("Did not identify a presubmit build. Exiting.")
	return 0

	if change_info:
	success = _invoke_using_change_info_json(
	dist_dir=dist_dir,
	change_info=change_info
	)
	else:
	# Fallback, since change-info is not guaranteed to exist
	success = _invoke_using_applied_prop(dist_dir=dist_dir)

	return 0 if success else 1


	if __name__ == "__main__":
	logging.basicConfig(level=_LOG_LEVEL,
	format="%(levelname)s: %(message)s")
	known, checkpatch_args = load_arguments()
	sys.exit(main(checkpatch_args=checkpatch_args, **vars(known)))