git_command.py - git-repo - Git at Google

 # Copyright (C) 2008 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.

 import functools
 import os
 import sys
 import subprocess
 from typing import Any, Optional

 from error import GitError
 from git_refs import HEAD
 import platform_utils
 from repo_trace import REPO_TRACE, IsTrace, Trace
 from wrapper import Wrapper

 GIT = "git"
 # NB: These do not need to be kept in sync with the repo launcher script.
 # These may be much newer as it allows the repo launcher to roll between
 # different repo releases while source versions might require a newer git.
 #
 # The soft version is when we start warning users that the version is old and
 # we'll be dropping support for it.  We'll refuse to work with versions older
 # than the hard version.
 #
 # git-1.7 is in (EOL) Ubuntu Precise.  git-1.9 is in Ubuntu Trusty.
 MIN_GIT_VERSION_SOFT = (1, 9, 1)
 MIN_GIT_VERSION_HARD = (1, 7, 2)
 GIT_DIR = "GIT_DIR"

 LAST_GITDIR = None
 LAST_CWD = None


 class _GitCall(object):
     @functools.lru_cache(maxsize=None)
     def version_tuple(self):
         ret = Wrapper().ParseGitVersion()
         if ret is None:
             print("fatal: unable to detect git version", file=sys.stderr)
             sys.exit(1)
         return ret

     def __getattr__(self, name):
         name = name.replace("_", "-")

         def fun(*cmdv):
             command = [name]
             command.extend(cmdv)
             return GitCommand(None, command).Wait() == 0

         return fun


 git = _GitCall()


 def RepoSourceVersion():
     """Return the version of the repo.git tree."""
     ver = getattr(RepoSourceVersion, "version", None)

     # We avoid GitCommand so we don't run into circular deps -- GitCommand needs
     # to initialize version info we provide.
     if ver is None:
         env = GitCommand._GetBasicEnv()

         proj = os.path.dirname(os.path.abspath(__file__))
         env[GIT_DIR] = os.path.join(proj, ".git")
         result = subprocess.run(
             [GIT, "describe", HEAD],
             stdout=subprocess.PIPE,
             stderr=subprocess.DEVNULL,
             encoding="utf-8",
             env=env,
             check=False,
         )
         if result.returncode == 0:
             ver = result.stdout.strip()
             if ver.startswith("v"):
                 ver = ver[1:]
         else:
             ver = "unknown"
         setattr(RepoSourceVersion, "version", ver)

     return ver


 class UserAgent(object):
     """Mange User-Agent settings when talking to external services

     We follow the style as documented here:
     https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent
     """

     _os = None
     _repo_ua = None
     _git_ua = None

     @property
     def os(self):
         """The operating system name."""
         if self._os is None:
             os_name = sys.platform
             if os_name.lower().startswith("linux"):
                 os_name = "Linux"
             elif os_name == "win32":
                 os_name = "Win32"
             elif os_name == "cygwin":
                 os_name = "Cygwin"
             elif os_name == "darwin":
                 os_name = "Darwin"
             self._os = os_name

         return self._os

     @property
     def repo(self):
         """The UA when connecting directly from repo."""
         if self._repo_ua is None:
             py_version = sys.version_info
             self._repo_ua = "git-repo/%s (%s) git/%s Python/%d.%d.%d" % (
                 RepoSourceVersion(),
                 self.os,
                 git.version_tuple().full,
                 py_version.major,
                 py_version.minor,
                 py_version.micro,
             )

         return self._repo_ua

     @property
     def git(self):
         """The UA when running git."""
         if self._git_ua is None:
             self._git_ua = "git/%s (%s) git-repo/%s" % (
                 git.version_tuple().full,
                 self.os,
                 RepoSourceVersion(),
             )

         return self._git_ua


 user_agent = UserAgent()


 def git_require(min_version, fail=False, msg=""):
     git_version = git.version_tuple()
     if min_version <= git_version:
         return True
     if fail:
         need = ".".join(map(str, min_version))
         if msg:
             msg = " for " + msg
         print(
             "fatal: git %s or later required%s" % (need, msg), file=sys.stderr
         )
         sys.exit(1)
     return False


 def _build_env(
     _kwargs_only=(),
     bare: Optional[bool] = False,
     disable_editor: Optional[bool] = False,
     ssh_proxy: Optional[Any] = None,
     gitdir: Optional[str] = None,
     objdir: Optional[str] = None,
 ):
     """Constucts an env dict for command execution."""

     assert _kwargs_only == (), "_build_env only accepts keyword arguments."

     env = GitCommand._GetBasicEnv()

     if disable_editor:
         env["GIT_EDITOR"] = ":"
     if ssh_proxy:
         env["REPO_SSH_SOCK"] = ssh_proxy.sock()
         env["GIT_SSH"] = ssh_proxy.proxy
         env["GIT_SSH_VARIANT"] = "ssh"
     if "http_proxy" in env and "darwin" == sys.platform:
         s = "'http.proxy=%s'" % (env["http_proxy"],)
         p = env.get("GIT_CONFIG_PARAMETERS")
         if p is not None:
             s = p + " " + s
         env["GIT_CONFIG_PARAMETERS"] = s
     if "GIT_ALLOW_PROTOCOL" not in env:
         env[
             "GIT_ALLOW_PROTOCOL"
         ] = "file:git:http:https:ssh:persistent-http:persistent-https:sso:rpc"
     env["GIT_HTTP_USER_AGENT"] = user_agent.git

     if objdir:
         # Set to the place we want to save the objects.
         env["GIT_OBJECT_DIRECTORY"] = objdir

         alt_objects = os.path.join(gitdir, "objects") if gitdir else None
         if alt_objects and os.path.realpath(alt_objects) != os.path.realpath(
             objdir
         ):
             # Allow git to search the original place in case of local or unique
             # refs that git will attempt to resolve even if we aren't fetching
             # them.
             env["GIT_ALTERNATE_OBJECT_DIRECTORIES"] = alt_objects
     if bare and gitdir is not None:
         env[GIT_DIR] = gitdir

     return env


 class GitCommand(object):
     """Wrapper around a single git invocation."""

     def __init__(
         self,
         project,
         cmdv,
         bare=False,
         input=None,
         capture_stdout=False,
         capture_stderr=False,
         merge_output=False,
         disable_editor=False,
         ssh_proxy=None,
         cwd=None,
         gitdir=None,
         objdir=None,
     ):
         if project:
             if not cwd:
                 cwd = project.worktree
             if not gitdir:
                 gitdir = project.gitdir

         # Git on Windows wants its paths only using / for reliability.
         if platform_utils.isWindows():
             if objdir:
                 objdir = objdir.replace("\\", "/")
             if gitdir:
                 gitdir = gitdir.replace("\\", "/")

         env = _build_env(
             disable_editor=disable_editor,
             ssh_proxy=ssh_proxy,
             objdir=objdir,
             gitdir=gitdir,
             bare=bare,
         )

         command = [GIT]
         if bare:
             cwd = None
         command.append(cmdv[0])
         # Need to use the --progress flag for fetch/clone so output will be
         # displayed as by default git only does progress output if stderr is a
         # TTY.
         if sys.stderr.isatty() and cmdv[0] in ("fetch", "clone"):
             if "--progress" not in cmdv and "--quiet" not in cmdv:
                 command.append("--progress")
         command.extend(cmdv[1:])

         stdin = subprocess.PIPE if input else None
         stdout = subprocess.PIPE if capture_stdout else None
         stderr = (
             subprocess.STDOUT
             if merge_output
             else (subprocess.PIPE if capture_stderr else None)
         )

         dbg = ""
         if IsTrace():
             global LAST_CWD
             global LAST_GITDIR

             if cwd and LAST_CWD != cwd:
                 if LAST_GITDIR or LAST_CWD:
                     dbg += "\n"
                 dbg += ": cd %s\n" % cwd
                 LAST_CWD = cwd

             if GIT_DIR in env and LAST_GITDIR != env[GIT_DIR]:
                 if LAST_GITDIR or LAST_CWD:
                     dbg += "\n"
                 dbg += ": export GIT_DIR=%s\n" % env[GIT_DIR]
                 LAST_GITDIR = env[GIT_DIR]

             if "GIT_OBJECT_DIRECTORY" in env:
                 dbg += (
                     ": export GIT_OBJECT_DIRECTORY=%s\n"
                     % env["GIT_OBJECT_DIRECTORY"]
                 )
             if "GIT_ALTERNATE_OBJECT_DIRECTORIES" in env:
                 dbg += ": export GIT_ALTERNATE_OBJECT_DIRECTORIES=%s\n" % (
                     env["GIT_ALTERNATE_OBJECT_DIRECTORIES"]
                 )

             dbg += ": "
             dbg += " ".join(command)
             if stdin == subprocess.PIPE:
                 dbg += " 0<|"
             if stdout == subprocess.PIPE:
                 dbg += " 1>|"
             if stderr == subprocess.PIPE:
                 dbg += " 2>|"
             elif stderr == subprocess.STDOUT:
                 dbg += " 2>&1"

         with Trace(
             "git command %s %s with debug: %s", LAST_GITDIR, command, dbg
         ):
             try:
                 p = subprocess.Popen(
                     command,
                     cwd=cwd,
                     env=env,
                     encoding="utf-8",
                     errors="backslashreplace",
                     stdin=stdin,
                     stdout=stdout,
                     stderr=stderr,
                 )
             except Exception as e:
                 raise GitError("%s: %s" % (command[1], e))

             if ssh_proxy:
                 ssh_proxy.add_client(p)

             self.process = p

             try:
                 self.stdout, self.stderr = p.communicate(input=input)
             finally:
                 if ssh_proxy:
                     ssh_proxy.remove_client(p)
             self.rc = p.wait()

     @staticmethod
     def _GetBasicEnv():
         """Return a basic env for running git under.

         This is guaranteed to be side-effect free.
         """
         env = os.environ.copy()
         for key in (
             REPO_TRACE,
             GIT_DIR,
             "GIT_ALTERNATE_OBJECT_DIRECTORIES",
             "GIT_OBJECT_DIRECTORY",
             "GIT_WORK_TREE",
             "GIT_GRAFT_FILE",
             "GIT_INDEX_FILE",
         ):
             env.pop(key, None)
         return env

     def Wait(self):
         return self.rc
	# Copyright (C) 2008 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.

	import functools
	import os
	import sys
	import subprocess
	from typing import Any, Optional

	from error import GitError
	from git_refs import HEAD
	import platform_utils
	from repo_trace import REPO_TRACE, IsTrace, Trace
	from wrapper import Wrapper

	GIT = "git"
	# NB: These do not need to be kept in sync with the repo launcher script.
	# These may be much newer as it allows the repo launcher to roll between
	# different repo releases while source versions might require a newer git.
	#
	# The soft version is when we start warning users that the version is old and
	# we'll be dropping support for it. We'll refuse to work with versions older
	# than the hard version.
	#
	# git-1.7 is in (EOL) Ubuntu Precise. git-1.9 is in Ubuntu Trusty.
	MIN_GIT_VERSION_SOFT = (1, 9, 1)
	MIN_GIT_VERSION_HARD = (1, 7, 2)
	GIT_DIR = "GIT_DIR"

	LAST_GITDIR = None
	LAST_CWD = None


	class _GitCall(object):
	@functools.lru_cache(maxsize=None)
	def version_tuple(self):
	ret = Wrapper().ParseGitVersion()
	if ret is None:
	print("fatal: unable to detect git version", file=sys.stderr)
	sys.exit(1)
	return ret

	def __getattr__(self, name):
	name = name.replace("_", "-")

	def fun(*cmdv):
	command = [name]
	command.extend(cmdv)
	return GitCommand(None, command).Wait() == 0

	return fun


	git = _GitCall()


	def RepoSourceVersion():
	"""Return the version of the repo.git tree."""
	ver = getattr(RepoSourceVersion, "version", None)

	# We avoid GitCommand so we don't run into circular deps -- GitCommand needs
	# to initialize version info we provide.
	if ver is None:
	env = GitCommand._GetBasicEnv()

	proj = os.path.dirname(os.path.abspath(__file__))
	env[GIT_DIR] = os.path.join(proj, ".git")
	result = subprocess.run(
	[GIT, "describe", HEAD],
	stdout=subprocess.PIPE,
	stderr=subprocess.DEVNULL,
	encoding="utf-8",
	env=env,
	check=False,
	)
	if result.returncode == 0:
	ver = result.stdout.strip()
	if ver.startswith("v"):
	ver = ver[1:]
	else:
	ver = "unknown"
	setattr(RepoSourceVersion, "version", ver)

	return ver


	class UserAgent(object):
	"""Mange User-Agent settings when talking to external services

	We follow the style as documented here:
	https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/User-Agent
	"""

	_os = None
	_repo_ua = None
	_git_ua = None

	@property
	def os(self):
	"""The operating system name."""
	if self._os is None:
	os_name = sys.platform
	if os_name.lower().startswith("linux"):
	os_name = "Linux"
	elif os_name == "win32":
	os_name = "Win32"
	elif os_name == "cygwin":
	os_name = "Cygwin"
	elif os_name == "darwin":
	os_name = "Darwin"
	self._os = os_name

	return self._os

	@property
	def repo(self):
	"""The UA when connecting directly from repo."""
	if self._repo_ua is None:
	py_version = sys.version_info
	self._repo_ua = "git-repo/%s (%s) git/%s Python/%d.%d.%d" % (
	RepoSourceVersion(),
	self.os,
	git.version_tuple().full,
	py_version.major,
	py_version.minor,
	py_version.micro,
	)

	return self._repo_ua

	@property
	def git(self):
	"""The UA when running git."""
	if self._git_ua is None:
	self._git_ua = "git/%s (%s) git-repo/%s" % (
	git.version_tuple().full,
	self.os,
	RepoSourceVersion(),
	)

	return self._git_ua


	user_agent = UserAgent()


	def git_require(min_version, fail=False, msg=""):
	git_version = git.version_tuple()
	if min_version <= git_version:
	return True
	if fail:
	need = ".".join(map(str, min_version))
	if msg:
	msg = " for " + msg
	print(
	"fatal: git %s or later required%s" % (need, msg), file=sys.stderr
	)
	sys.exit(1)
	return False


	def _build_env(
	_kwargs_only=(),
	bare: Optional[bool] = False,
	disable_editor: Optional[bool] = False,
	ssh_proxy: Optional[Any] = None,
	gitdir: Optional[str] = None,
	objdir: Optional[str] = None,
	):
	"""Constucts an env dict for command execution."""

	assert _kwargs_only == (), "_build_env only accepts keyword arguments."

	env = GitCommand._GetBasicEnv()

	if disable_editor:
	env["GIT_EDITOR"] = ":"
	if ssh_proxy:
	env["REPO_SSH_SOCK"] = ssh_proxy.sock()
	env["GIT_SSH"] = ssh_proxy.proxy
	env["GIT_SSH_VARIANT"] = "ssh"
	if "http_proxy" in env and "darwin" == sys.platform:
	s = "'http.proxy=%s'" % (env["http_proxy"],)
	p = env.get("GIT_CONFIG_PARAMETERS")
	if p is not None:
	s = p + " " + s
	env["GIT_CONFIG_PARAMETERS"] = s
	if "GIT_ALLOW_PROTOCOL" not in env:
	env[
	"GIT_ALLOW_PROTOCOL"
	] = "file:git:http:https:ssh:persistent-http:persistent-https:sso:rpc"
	env["GIT_HTTP_USER_AGENT"] = user_agent.git

	if objdir:
	# Set to the place we want to save the objects.
	env["GIT_OBJECT_DIRECTORY"] = objdir

	alt_objects = os.path.join(gitdir, "objects") if gitdir else None
	if alt_objects and os.path.realpath(alt_objects) != os.path.realpath(
	objdir
	):
	# Allow git to search the original place in case of local or unique
	# refs that git will attempt to resolve even if we aren't fetching
	# them.
	env["GIT_ALTERNATE_OBJECT_DIRECTORIES"] = alt_objects
	if bare and gitdir is not None:
	env[GIT_DIR] = gitdir

	return env


	class GitCommand(object):
	"""Wrapper around a single git invocation."""

	def __init__(
	self,
	project,
	cmdv,
	bare=False,
	input=None,
	capture_stdout=False,
	capture_stderr=False,
	merge_output=False,
	disable_editor=False,
	ssh_proxy=None,
	cwd=None,
	gitdir=None,
	objdir=None,
	):
	if project:
	if not cwd:
	cwd = project.worktree
	if not gitdir:
	gitdir = project.gitdir

	# Git on Windows wants its paths only using / for reliability.
	if platform_utils.isWindows():
	if objdir:
	objdir = objdir.replace("\\", "/")
	if gitdir:
	gitdir = gitdir.replace("\\", "/")

	env = _build_env(
	disable_editor=disable_editor,
	ssh_proxy=ssh_proxy,
	objdir=objdir,
	gitdir=gitdir,
	bare=bare,
	)

	command = [GIT]
	if bare:
	cwd = None
	command.append(cmdv[0])
	# Need to use the --progress flag for fetch/clone so output will be
	# displayed as by default git only does progress output if stderr is a
	# TTY.
	if sys.stderr.isatty() and cmdv[0] in ("fetch", "clone"):
	if "--progress" not in cmdv and "--quiet" not in cmdv:
	command.append("--progress")
	command.extend(cmdv[1:])

	stdin = subprocess.PIPE if input else None
	stdout = subprocess.PIPE if capture_stdout else None
	stderr = (
	subprocess.STDOUT
	if merge_output
	else (subprocess.PIPE if capture_stderr else None)
	)

	dbg = ""
	if IsTrace():
	global LAST_CWD
	global LAST_GITDIR

	if cwd and LAST_CWD != cwd:
	if LAST_GITDIR or LAST_CWD:
	dbg += "\n"
	dbg += ": cd %s\n" % cwd
	LAST_CWD = cwd

	if GIT_DIR in env and LAST_GITDIR != env[GIT_DIR]:
	if LAST_GITDIR or LAST_CWD:
	dbg += "\n"
	dbg += ": export GIT_DIR=%s\n" % env[GIT_DIR]
	LAST_GITDIR = env[GIT_DIR]

	if "GIT_OBJECT_DIRECTORY" in env:
	dbg += (
	": export GIT_OBJECT_DIRECTORY=%s\n"
	% env["GIT_OBJECT_DIRECTORY"]
	)
	if "GIT_ALTERNATE_OBJECT_DIRECTORIES" in env:
	dbg += ": export GIT_ALTERNATE_OBJECT_DIRECTORIES=%s\n" % (
	env["GIT_ALTERNATE_OBJECT_DIRECTORIES"]
	)

	dbg += ": "
	dbg += " ".join(command)
	if stdin == subprocess.PIPE:
	dbg += " 0<\|"
	if stdout == subprocess.PIPE:
	dbg += " 1>\|"
	if stderr == subprocess.PIPE:
	dbg += " 2>\|"
	elif stderr == subprocess.STDOUT:
	dbg += " 2>&1"

	with Trace(
	"git command %s %s with debug: %s", LAST_GITDIR, command, dbg
	):
	try:
	p = subprocess.Popen(
	command,
	cwd=cwd,
	env=env,
	encoding="utf-8",
	errors="backslashreplace",
	stdin=stdin,
	stdout=stdout,
	stderr=stderr,
	)
	except Exception as e:
	raise GitError("%s: %s" % (command[1], e))

	if ssh_proxy:
	ssh_proxy.add_client(p)

	self.process = p

	try:
	self.stdout, self.stderr = p.communicate(input=input)
	finally:
	if ssh_proxy:
	ssh_proxy.remove_client(p)
	self.rc = p.wait()

	@staticmethod
	def _GetBasicEnv():
	"""Return a basic env for running git under.

	This is guaranteed to be side-effect free.
	"""
	env = os.environ.copy()
	for key in (
	REPO_TRACE,
	GIT_DIR,
	"GIT_ALTERNATE_OBJECT_DIRECTORIES",
	"GIT_OBJECT_DIRECTORY",
	"GIT_WORK_TREE",
	"GIT_GRAFT_FILE",
	"GIT_INDEX_FILE",
	):
	env.pop(key, None)
	return env

	def Wait(self):
	return self.rc