pw_presubmit/py/pw_presubmit/git_repo.py - platform/external/pigweed - Git at Google

 # Copyright 2020 The Pigweed Authors
 #
 # 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
 #
 #     https://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.
 """Helpful commands for working with a Git repository."""

 import logging
 from pathlib import Path
 import subprocess
 from typing import Collection, Iterable, List, Optional, Pattern, Union

 from pw_presubmit.tools import log_run, plural

 _LOG = logging.getLogger(__name__)
 PathOrStr = Union[Path, str]
 PatternOrStr = Union[Pattern, str]

 TRACKING_BRANCH_ALIAS = '@{upstream}'
 _TRACKING_BRANCH_ALIASES = TRACKING_BRANCH_ALIAS, '@{u}'


 def git_stdout(*args: PathOrStr,
                show_stderr=False,
                repo: PathOrStr = '.') -> str:
     return log_run(['git', '-C', str(repo), *args],
                    stdout=subprocess.PIPE,
                    stderr=None if show_stderr else subprocess.DEVNULL,
                    check=True).stdout.decode().strip()


 def _ls_files(args: Collection[PathOrStr], repo: Path) -> Iterable[Path]:
     """Returns results of git ls-files as absolute paths."""
     git_root = repo.resolve()
     for file in git_stdout('ls-files', '--', *args, repo=repo).splitlines():
         full_path = git_root / file
         # Modified submodules will show up as directories and should be ignored.
         if full_path.is_file():
             yield full_path


 def _diff_names(commit: str, pathspecs: Collection[PathOrStr],
                 repo: Path) -> Iterable[Path]:
     """Returns absolute paths of files changed since the specified commit."""
     git_root = root(repo)
     for file in git_stdout('diff',
                            '--name-only',
                            '--diff-filter=d',
                            commit,
                            '--',
                            *pathspecs,
                            repo=repo).splitlines():
         full_path = git_root / file
         # Modified submodules will show up as directories and should be ignored.
         if full_path.is_file():
             yield full_path


 def tracking_branch(repo_path: Optional[Path] = None) -> Optional[str]:
     """Returns the tracking branch of the current branch.

     Since most callers of this function can safely handle a return value of
     None, suppress exceptions and return None if there is no tracking branch.

     Args:
       repo_path: repo path from which to run commands; defaults to Path.cwd()

     Raises:
       ValueError: if repo_path is not in a Git repository

     Returns:
       the remote tracking branch name or None if there is none
     """
     if repo_path is None:
         repo_path = Path.cwd()

     if not is_repo(repo_path or Path.cwd()):
         raise ValueError(f'{repo_path} is not within a Git repository')

     # This command should only error out if there's no upstream branch set.
     try:
         return git_stdout('rev-parse',
                           '--abbrev-ref',
                           '--symbolic-full-name',
                           TRACKING_BRANCH_ALIAS,
                           repo=repo_path)

     except subprocess.CalledProcessError:
         return None


 def list_files(commit: Optional[str] = None,
                pathspecs: Collection[PathOrStr] = (),
                repo_path: Optional[Path] = None) -> List[Path]:
     """Lists files with git ls-files or git diff --name-only.

     Args:
       commit: commit to use as a base for git diff
       pathspecs: Git pathspecs to use in git ls-files or diff
       repo_path: repo path from which to run commands; defaults to Path.cwd()

     Returns:
       A sorted list of absolute paths
     """
     if repo_path is None:
         repo_path = Path.cwd()

     if commit in _TRACKING_BRANCH_ALIASES:
         commit = tracking_branch(repo_path)

     if commit:
         try:
             return sorted(_diff_names(commit, pathspecs, repo_path))
         except subprocess.CalledProcessError:
             _LOG.warning(
                 'Error comparing with base revision %s of %s, listing all '
                 'files instead of just changed files', commit, repo_path)

     return sorted(_ls_files(pathspecs, repo_path))


 def has_uncommitted_changes(repo: Optional[Path] = None) -> bool:
     """Returns True if the Git repo has uncommitted changes in it.

     This does not check for untracked files.
     """
     if repo is None:
         repo = Path.cwd()

     # Refresh the Git index so that the diff-index command will be accurate.
     # The `git update-index` command isn't reliable when run in parallel with
     # other processes that may touch files in the repo directory, so retry a
     # few times before giving up. The hallmark of this failure mode is the lack
     # of an error message on stderr, so if we see something there we can assume
     # it's some other issue and raise.
     retries = 6
     for i in range(retries):
         try:
             log_run(['git', '-C', repo, 'update-index', '-q', '--refresh'],
                     capture_output=True,
                     check=True)
         except subprocess.CalledProcessError as err:
             if err.stderr or i == retries - 1:
                 raise
             continue
     # diff-index exits with 1 if there are uncommitted changes.
     return log_run(['git', '-C', repo, 'diff-index', '--quiet', 'HEAD',
                     '--']).returncode == 1


 def _describe_constraints(git_root: Path, repo_path: Path,
                           commit: Optional[str],
                           pathspecs: Collection[PathOrStr],
                           exclude: Collection[Pattern[str]]) -> Iterable[str]:
     if not git_root.samefile(repo_path):
         yield (
             f'under the {repo_path.resolve().relative_to(git_root.resolve())} '
             'subdirectory')

     if commit in _TRACKING_BRANCH_ALIASES:
         commit = tracking_branch(git_root)
         if commit is None:
             _LOG.warning(
                 'Attempted to list files changed since the remote tracking '
                 'branch, but the repo is not tracking a branch')

     if commit:
         yield f'that have changed since {commit}'

     if pathspecs:
         paths_str = ', '.join(str(p) for p in pathspecs)
         yield f'that match {plural(pathspecs, "pathspec")} ({paths_str})'

     if exclude:
         yield (f'that do not match {plural(exclude, "pattern")} (' +
                ', '.join(p.pattern for p in exclude) + ')')


 def describe_files(git_root: Path,
                    repo_path: Path,
                    commit: Optional[str],
                    pathspecs: Collection[PathOrStr],
                    exclude: Collection[Pattern],
                    project_root: Optional[Path] = None) -> str:
     """Completes 'Doing something to ...' for a set of files in a Git repo."""
     constraints = list(
         _describe_constraints(git_root, repo_path, commit, pathspecs, exclude))

     name = git_root.name
     if project_root and project_root != git_root:
         name = str(git_root.relative_to(project_root))

     if not constraints:
         return f'all files in the {name} repo'

     msg = f'files in the {name} repo'
     if len(constraints) == 1:
         return f'{msg} {constraints[0]}'

     return msg + ''.join(f'\n    - {line}' for line in constraints)


 def root(repo_path: PathOrStr = '.', *, show_stderr: bool = True) -> Path:
     """Returns the repository root as an absolute path.

     Raises:
       FileNotFoundError: the path does not exist
       subprocess.CalledProcessError: the path is not in a Git repo
     """
     repo_path = Path(repo_path)
     if not repo_path.exists():
         raise FileNotFoundError(f'{repo_path} does not exist')

     return Path(
         git_stdout('rev-parse',
                    '--show-toplevel',
                    repo=repo_path if repo_path.is_dir() else repo_path.parent,
                    show_stderr=show_stderr))


 def within_repo(repo_path: PathOrStr = '.') -> Optional[Path]:
     """Similar to root(repo_path), returns None if the path is not in a repo."""
     try:
         return root(repo_path, show_stderr=False)
     except subprocess.CalledProcessError:
         return None


 def is_repo(repo_path: PathOrStr = '.') -> bool:
     """True if the path is tracked by a Git repo."""
     return within_repo(repo_path) is not None


 def path(repo_path: PathOrStr,
          *additional_repo_paths: PathOrStr,
          repo: PathOrStr = '.') -> Path:
     """Returns a path relative to a Git repository's root."""
     return root(repo).joinpath(repo_path, *additional_repo_paths)


 def commit_message(commit: str = 'HEAD', repo: PathOrStr = '.') -> str:
     return git_stdout('log', '--format=%B', '-n1', commit, repo=repo)


 def commit_author(commit: str = 'HEAD', repo: PathOrStr = '.') -> str:
     return git_stdout('log', '--format=%ae', '-n1', commit, repo=repo)


 def commit_hash(rev: str = 'HEAD',
                 short: bool = True,
                 repo: PathOrStr = '.') -> str:
     """Returns the commit hash of the revision."""
     args = ['rev-parse']
     if short:
         args += ['--short']
     args += [rev]
     return git_stdout(*args, repo=repo)


 def discover_submodules(
     superproject_dir: Path, excluded_paths: Collection[PatternOrStr] = ()
 ) -> List[Path]:
     """Query git and return a list of submodules in the current project.

     Args:
         superproject_dir: Path object to directory under which we are looking
                           for submodules. This will also be included in list
                           returned unless excluded.
         excluded_paths: Pattern or string that match submodules that should not
                         be returned. All matches are done on posix style paths.

     Returns:
         List of "Path"s which were found but not excluded, this includes
         superproject_dir unless excluded.
     """
     discovery_report = git_stdout('submodule',
                                   'foreach',
                                   '--quiet',
                                   '--recursive',
                                   'echo $toplevel/$sm_path',
                                   repo=superproject_dir)
     module_dirs = [Path(line) for line in discovery_report.split()]
     # The superproject is omitted in the prior scan.
     module_dirs.append(superproject_dir)

     for exclude in excluded_paths:
         if isinstance(exclude, Pattern):
             for module_dir in reversed(module_dirs):
                 if exclude.fullmatch(module_dir.as_posix()):
                     module_dirs.remove(module_dir)
         else:
             for module_dir in reversed(module_dirs):
                 if exclude == module_dir.as_posix():
                     module_dirs.remove(module_dir)

     return module_dirs
	# Copyright 2020 The Pigweed Authors
	#
	# 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
	#
	# https://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.
	"""Helpful commands for working with a Git repository."""

	import logging
	from pathlib import Path
	import subprocess
	from typing import Collection, Iterable, List, Optional, Pattern, Union

	from pw_presubmit.tools import log_run, plural

	_LOG = logging.getLogger(__name__)
	PathOrStr = Union[Path, str]
	PatternOrStr = Union[Pattern, str]

	TRACKING_BRANCH_ALIAS = '@{upstream}'
	_TRACKING_BRANCH_ALIASES = TRACKING_BRANCH_ALIAS, '@{u}'


	def git_stdout(*args: PathOrStr,
	show_stderr=False,
	repo: PathOrStr = '.') -> str:
	return log_run(['git', '-C', str(repo), *args],
	stdout=subprocess.PIPE,
	stderr=None if show_stderr else subprocess.DEVNULL,
	check=True).stdout.decode().strip()


	def _ls_files(args: Collection[PathOrStr], repo: Path) -> Iterable[Path]:
	"""Returns results of git ls-files as absolute paths."""
	git_root = repo.resolve()
	for file in git_stdout('ls-files', '--', *args, repo=repo).splitlines():
	full_path = git_root / file
	# Modified submodules will show up as directories and should be ignored.
	if full_path.is_file():
	yield full_path


	def _diff_names(commit: str, pathspecs: Collection[PathOrStr],
	repo: Path) -> Iterable[Path]:
	"""Returns absolute paths of files changed since the specified commit."""
	git_root = root(repo)
	for file in git_stdout('diff',
	'--name-only',
	'--diff-filter=d',
	commit,
	'--',
	*pathspecs,
	repo=repo).splitlines():
	full_path = git_root / file
	# Modified submodules will show up as directories and should be ignored.
	if full_path.is_file():
	yield full_path


	def tracking_branch(repo_path: Optional[Path] = None) -> Optional[str]:
	"""Returns the tracking branch of the current branch.

	Since most callers of this function can safely handle a return value of
	None, suppress exceptions and return None if there is no tracking branch.

	Args:
	repo_path: repo path from which to run commands; defaults to Path.cwd()

	Raises:
	ValueError: if repo_path is not in a Git repository

	Returns:
	the remote tracking branch name or None if there is none
	"""
	if repo_path is None:
	repo_path = Path.cwd()

	if not is_repo(repo_path or Path.cwd()):
	raise ValueError(f'{repo_path} is not within a Git repository')

	# This command should only error out if there's no upstream branch set.
	try:
	return git_stdout('rev-parse',
	'--abbrev-ref',
	'--symbolic-full-name',
	TRACKING_BRANCH_ALIAS,
	repo=repo_path)

	except subprocess.CalledProcessError:
	return None


	def list_files(commit: Optional[str] = None,
	pathspecs: Collection[PathOrStr] = (),
	repo_path: Optional[Path] = None) -> List[Path]:
	"""Lists files with git ls-files or git diff --name-only.

	Args:
	commit: commit to use as a base for git diff
	pathspecs: Git pathspecs to use in git ls-files or diff
	repo_path: repo path from which to run commands; defaults to Path.cwd()

	Returns:
	A sorted list of absolute paths
	"""
	if repo_path is None:
	repo_path = Path.cwd()

	if commit in _TRACKING_BRANCH_ALIASES:
	commit = tracking_branch(repo_path)

	if commit:
	try:
	return sorted(_diff_names(commit, pathspecs, repo_path))
	except subprocess.CalledProcessError:
	_LOG.warning(
	'Error comparing with base revision %s of %s, listing all '
	'files instead of just changed files', commit, repo_path)

	return sorted(_ls_files(pathspecs, repo_path))


	def has_uncommitted_changes(repo: Optional[Path] = None) -> bool:
	"""Returns True if the Git repo has uncommitted changes in it.

	This does not check for untracked files.
	"""
	if repo is None:
	repo = Path.cwd()

	# Refresh the Git index so that the diff-index command will be accurate.
	# The `git update-index` command isn't reliable when run in parallel with
	# other processes that may touch files in the repo directory, so retry a
	# few times before giving up. The hallmark of this failure mode is the lack
	# of an error message on stderr, so if we see something there we can assume
	# it's some other issue and raise.
	retries = 6
	for i in range(retries):
	try:
	log_run(['git', '-C', repo, 'update-index', '-q', '--refresh'],
	capture_output=True,
	check=True)
	except subprocess.CalledProcessError as err:
	if err.stderr or i == retries - 1:
	raise
	continue
	# diff-index exits with 1 if there are uncommitted changes.
	return log_run(['git', '-C', repo, 'diff-index', '--quiet', 'HEAD',
	'--']).returncode == 1


	def _describe_constraints(git_root: Path, repo_path: Path,
	commit: Optional[str],
	pathspecs: Collection[PathOrStr],
	exclude: Collection[Pattern[str]]) -> Iterable[str]:
	if not git_root.samefile(repo_path):
	yield (
	f'under the {repo_path.resolve().relative_to(git_root.resolve())} '
	'subdirectory')

	if commit in _TRACKING_BRANCH_ALIASES:
	commit = tracking_branch(git_root)
	if commit is None:
	_LOG.warning(
	'Attempted to list files changed since the remote tracking '
	'branch, but the repo is not tracking a branch')

	if commit:
	yield f'that have changed since {commit}'

	if pathspecs:
	paths_str = ', '.join(str(p) for p in pathspecs)
	yield f'that match {plural(pathspecs, "pathspec")} ({paths_str})'

	if exclude:
	yield (f'that do not match {plural(exclude, "pattern")} (' +
	', '.join(p.pattern for p in exclude) + ')')


	def describe_files(git_root: Path,
	repo_path: Path,
	commit: Optional[str],
	pathspecs: Collection[PathOrStr],
	exclude: Collection[Pattern],
	project_root: Optional[Path] = None) -> str:
	"""Completes 'Doing something to ...' for a set of files in a Git repo."""
	constraints = list(
	_describe_constraints(git_root, repo_path, commit, pathspecs, exclude))

	name = git_root.name
	if project_root and project_root != git_root:
	name = str(git_root.relative_to(project_root))

	if not constraints:
	return f'all files in the {name} repo'

	msg = f'files in the {name} repo'
	if len(constraints) == 1:
	return f'{msg} {constraints[0]}'

	return msg + ''.join(f'\n - {line}' for line in constraints)


	def root(repo_path: PathOrStr = '.', *, show_stderr: bool = True) -> Path:
	"""Returns the repository root as an absolute path.

	Raises:
	FileNotFoundError: the path does not exist
	subprocess.CalledProcessError: the path is not in a Git repo
	"""
	repo_path = Path(repo_path)
	if not repo_path.exists():
	raise FileNotFoundError(f'{repo_path} does not exist')

	return Path(
	git_stdout('rev-parse',
	'--show-toplevel',
	repo=repo_path if repo_path.is_dir() else repo_path.parent,
	show_stderr=show_stderr))


	def within_repo(repo_path: PathOrStr = '.') -> Optional[Path]:
	"""Similar to root(repo_path), returns None if the path is not in a repo."""
	try:
	return root(repo_path, show_stderr=False)
	except subprocess.CalledProcessError:
	return None


	def is_repo(repo_path: PathOrStr = '.') -> bool:
	"""True if the path is tracked by a Git repo."""
	return within_repo(repo_path) is not None


	def path(repo_path: PathOrStr,
	*additional_repo_paths: PathOrStr,
	repo: PathOrStr = '.') -> Path:
	"""Returns a path relative to a Git repository's root."""
	return root(repo).joinpath(repo_path, *additional_repo_paths)


	def commit_message(commit: str = 'HEAD', repo: PathOrStr = '.') -> str:
	return git_stdout('log', '--format=%B', '-n1', commit, repo=repo)


	def commit_author(commit: str = 'HEAD', repo: PathOrStr = '.') -> str:
	return git_stdout('log', '--format=%ae', '-n1', commit, repo=repo)


	def commit_hash(rev: str = 'HEAD',
	short: bool = True,
	repo: PathOrStr = '.') -> str:
	"""Returns the commit hash of the revision."""
	args = ['rev-parse']
	if short:
	args += ['--short']
	args += [rev]
	return git_stdout(*args, repo=repo)


	def discover_submodules(
	superproject_dir: Path, excluded_paths: Collection[PatternOrStr] = ()
	) -> List[Path]:
	"""Query git and return a list of submodules in the current project.

	Args:
	superproject_dir: Path object to directory under which we are looking
	for submodules. This will also be included in list
	returned unless excluded.
	excluded_paths: Pattern or string that match submodules that should not
	be returned. All matches are done on posix style paths.

	Returns:
	List of "Path"s which were found but not excluded, this includes
	superproject_dir unless excluded.
	"""
	discovery_report = git_stdout('submodule',
	'foreach',
	'--quiet',
	'--recursive',
	'echo $toplevel/$sm_path',
	repo=superproject_dir)
	module_dirs = [Path(line) for line in discovery_report.split()]
	# The superproject is omitted in the prior scan.
	module_dirs.append(superproject_dir)

	for exclude in excluded_paths:
	if isinstance(exclude, Pattern):
	for module_dir in reversed(module_dirs):
	if exclude.fullmatch(module_dir.as_posix()):
	module_dirs.remove(module_dir)
	else:
	for module_dir in reversed(module_dirs):
	if exclude == module_dir.as_posix():
	module_dirs.remove(module_dir)

	return module_dirs