blob: fca0e6f76cc13d45bea490732b3d203de1ad006d [file] [log] [blame]
# Copyright (C) 2021 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.
"""Provide functionality to get projects and their commit ids from Superproject.
For more information on superproject, check out:
https://en.wikibooks.org/wiki/Git/Submodules_and_Superprojects
Examples:
superproject = Superproject(manifest, name, remote, revision)
UpdateProjectsResult = superproject.UpdateProjectsRevisionId(projects)
"""
import functools
import hashlib
import os
import sys
import time
from typing import NamedTuple
from git_command import git_require
from git_command import GitCommand
from git_config import RepoConfig
from git_refs import GitRefs
_SUPERPROJECT_GIT_NAME = "superproject.git"
_SUPERPROJECT_MANIFEST_NAME = "superproject_override.xml"
class SyncResult(NamedTuple):
"""Return the status of sync and whether caller should exit."""
# Whether the superproject sync was successful.
success: bool
# Whether the caller should exit.
fatal: bool
class CommitIdsResult(NamedTuple):
"""Return the commit ids and whether caller should exit."""
# A dictionary with the projects/commit ids on success, otherwise None.
commit_ids: dict
# Whether the caller should exit.
fatal: bool
class UpdateProjectsResult(NamedTuple):
"""Return the overriding manifest file and whether caller should exit."""
# Path name of the overriding manifest file if successful, otherwise None.
manifest_path: str
# Whether the caller should exit.
fatal: bool
class Superproject:
"""Get commit ids from superproject.
Initializes a bare local copy of a superproject for the manifest. This
allows lookup of commit ids for all projects. It contains
_project_commit_ids which is a dictionary with project/commit id entries.
"""
def __init__(
self,
manifest,
name,
remote,
revision,
superproject_dir="exp-superproject",
):
"""Initializes superproject.
Args:
manifest: A Manifest object that is to be written to a file.
name: The unique name of the superproject
remote: The RemoteSpec for the remote.
revision: The name of the git branch to track.
superproject_dir: Relative path under |manifest.subdir| to checkout
superproject.
"""
self._project_commit_ids = None
self._manifest = manifest
self.name = name
self.remote = remote
self.revision = self._branch = revision
self._repodir = manifest.repodir
self._superproject_dir = superproject_dir
self._superproject_path = manifest.SubmanifestInfoDir(
manifest.path_prefix, superproject_dir
)
self._manifest_path = os.path.join(
self._superproject_path, _SUPERPROJECT_MANIFEST_NAME
)
git_name = hashlib.md5(remote.name.encode("utf8")).hexdigest() + "-"
self._remote_url = remote.url
self._work_git_name = git_name + _SUPERPROJECT_GIT_NAME
self._work_git = os.path.join(
self._superproject_path, self._work_git_name
)
# The following are command arguemnts, rather than superproject
# attributes, and were included here originally. They should eventually
# become arguments that are passed down from the public methods, instead
# of being treated as attributes.
self._git_event_log = None
self._quiet = False
self._print_messages = False
def SetQuiet(self, value):
"""Set the _quiet attribute."""
self._quiet = value
def SetPrintMessages(self, value):
"""Set the _print_messages attribute."""
self._print_messages = value
@property
def project_commit_ids(self):
"""Returns a dictionary of projects and their commit ids."""
return self._project_commit_ids
@property
def manifest_path(self):
"""Returns the manifest path if the path exists or None."""
return (
self._manifest_path if os.path.exists(self._manifest_path) else None
)
def _LogMessage(self, fmt, *inputs):
"""Logs message to stderr and _git_event_log."""
message = f"{self._LogMessagePrefix()} {fmt.format(*inputs)}"
if self._print_messages:
print(message, file=sys.stderr)
self._git_event_log.ErrorEvent(message, fmt)
def _LogMessagePrefix(self):
"""Returns the prefix string to be logged in each log message"""
return (
f"repo superproject branch: {self._branch} url: {self._remote_url}"
)
def _LogError(self, fmt, *inputs):
"""Logs error message to stderr and _git_event_log."""
self._LogMessage(f"error: {fmt}", *inputs)
def _LogWarning(self, fmt, *inputs):
"""Logs warning message to stderr and _git_event_log."""
self._LogMessage(f"warning: {fmt}", *inputs)
def _Init(self):
"""Sets up a local Git repository to get a copy of a superproject.
Returns:
True if initialization is successful, or False.
"""
if not os.path.exists(self._superproject_path):
os.mkdir(self._superproject_path)
if not self._quiet and not os.path.exists(self._work_git):
print(
"%s: Performing initial setup for superproject; this might "
"take several minutes." % self._work_git
)
cmd = ["init", "--bare", self._work_git_name]
p = GitCommand(
None,
cmd,
cwd=self._superproject_path,
capture_stdout=True,
capture_stderr=True,
)
retval = p.Wait()
if retval:
self._LogWarning(
"git init call failed, command: git {}, "
"return code: {}, stderr: {}",
cmd,
retval,
p.stderr,
)
return False
return True
def _Fetch(self):
"""Fetches a superproject for the manifest based on |_remote_url|.
This runs git fetch which stores a local copy the superproject.
Returns:
True if fetch is successful, or False.
"""
if not os.path.exists(self._work_git):
self._LogWarning("git fetch missing directory: {}", self._work_git)
return False
if not git_require((2, 28, 0)):
self._LogWarning(
"superproject requires a git version 2.28 or later"
)
return False
cmd = [
"fetch",
self._remote_url,
"--depth",
"1",
"--force",
"--no-tags",
"--filter",
"blob:none",
]
# Check if there is a local ref that we can pass to --negotiation-tip.
# If this is the first fetch, it does not exist yet.
# We use --negotiation-tip to speed up the fetch. Superproject branches
# do not share commits. So this lets git know it only needs to send
# commits reachable from the specified local refs.
rev_commit = GitRefs(self._work_git).get(f"refs/heads/{self.revision}")
if rev_commit:
cmd.extend(["--negotiation-tip", rev_commit])
if self._branch:
cmd += [self._branch + ":" + self._branch]
p = GitCommand(
None,
cmd,
gitdir=self._work_git,
bare=True,
capture_stdout=True,
capture_stderr=True,
)
retval = p.Wait()
if retval:
self._LogWarning(
"git fetch call failed, command: git {}, "
"return code: {}, stderr: {}",
cmd,
retval,
p.stderr,
)
return False
return True
def _LsTree(self):
"""Gets the commit ids for all projects.
Works only in git repositories.
Returns:
data: data returned from 'git ls-tree ...' instead of None.
"""
if not os.path.exists(self._work_git):
self._LogWarning(
"git ls-tree missing directory: {}", self._work_git
)
return None
data = None
branch = "HEAD" if not self._branch else self._branch
cmd = ["ls-tree", "-z", "-r", branch]
p = GitCommand(
None,
cmd,
gitdir=self._work_git,
bare=True,
capture_stdout=True,
capture_stderr=True,
)
retval = p.Wait()
if retval == 0:
data = p.stdout
else:
self._LogWarning(
"git ls-tree call failed, command: git {}, "
"return code: {}, stderr: {}",
cmd,
retval,
p.stderr,
)
return data
def Sync(self, git_event_log):
"""Gets a local copy of a superproject for the manifest.
Args:
git_event_log: an EventLog, for git tracing.
Returns:
SyncResult
"""
self._git_event_log = git_event_log
if not self._manifest.superproject:
self._LogWarning(
"superproject tag is not defined in manifest: {}",
self._manifest.manifestFile,
)
return SyncResult(False, False)
_PrintBetaNotice()
should_exit = True
if not self._remote_url:
self._LogWarning(
"superproject URL is not defined in manifest: {}",
self._manifest.manifestFile,
)
return SyncResult(False, should_exit)
if not self._Init():
return SyncResult(False, should_exit)
if not self._Fetch():
return SyncResult(False, should_exit)
if not self._quiet:
print(
"%s: Initial setup for superproject completed." % self._work_git
)
return SyncResult(True, False)
def _GetAllProjectsCommitIds(self):
"""Get commit ids for all projects from superproject and save them.
Commit ids are saved in _project_commit_ids.
Returns:
CommitIdsResult
"""
sync_result = self.Sync(self._git_event_log)
if not sync_result.success:
return CommitIdsResult(None, sync_result.fatal)
data = self._LsTree()
if not data:
self._LogWarning(
"git ls-tree failed to return data for manifest: {}",
self._manifest.manifestFile,
)
return CommitIdsResult(None, True)
# Parse lines like the following to select lines starting with '160000'
# and build a dictionary with project path (last element) and its commit
# id (3rd element).
#
# 160000 commit 2c2724cb36cd5a9cec6c852c681efc3b7c6b86ea\tart\x00
# 120000 blob acc2cbdf438f9d2141f0ae424cec1d8fc4b5d97f\tbootstrap.bash\x00 # noqa: E501
commit_ids = {}
for line in data.split("\x00"):
ls_data = line.split(None, 3)
if not ls_data:
break
if ls_data[0] == "160000":
commit_ids[ls_data[3]] = ls_data[2]
self._project_commit_ids = commit_ids
return CommitIdsResult(commit_ids, False)
def _WriteManifestFile(self):
"""Writes manifest to a file.
Returns:
manifest_path: Path name of the file into which manifest is written
instead of None.
"""
if not os.path.exists(self._superproject_path):
self._LogWarning(
"missing superproject directory: {}", self._superproject_path
)
return None
manifest_str = self._manifest.ToXml(
groups=self._manifest.GetGroupsStr(), omit_local=True
).toxml()
manifest_path = self._manifest_path
try:
with open(manifest_path, "w", encoding="utf-8") as fp:
fp.write(manifest_str)
except OSError as e:
self._LogError("cannot write manifest to : {} {}", manifest_path, e)
return None
return manifest_path
def _SkipUpdatingProjectRevisionId(self, project):
"""Checks if a project's revision id needs to be updated or not.
Revision id for projects from local manifest will not be updated.
Args:
project: project whose revision id is being updated.
Returns:
True if a project's revision id should not be updated, or False,
"""
path = project.relpath
if not path:
return True
# Skip the project with revisionId.
if project.revisionId:
return True
# Skip the project if it comes from the local manifest.
return project.manifest.IsFromLocalManifest(project)
def UpdateProjectsRevisionId(self, projects, git_event_log):
"""Update revisionId of every project in projects with the commit id.
Args:
projects: a list of projects whose revisionId needs to be updated.
git_event_log: an EventLog, for git tracing.
Returns:
UpdateProjectsResult
"""
self._git_event_log = git_event_log
commit_ids_result = self._GetAllProjectsCommitIds()
commit_ids = commit_ids_result.commit_ids
if not commit_ids:
return UpdateProjectsResult(None, commit_ids_result.fatal)
projects_missing_commit_ids = []
for project in projects:
if self._SkipUpdatingProjectRevisionId(project):
continue
path = project.relpath
commit_id = commit_ids.get(path)
if not commit_id:
projects_missing_commit_ids.append(path)
# If superproject doesn't have a commit id for a project, then report an
# error event and continue as if do not use superproject is specified.
if projects_missing_commit_ids:
self._LogWarning(
"please file a bug using {} to report missing "
"commit_ids for: {}",
self._manifest.contactinfo.bugurl,
projects_missing_commit_ids,
)
return UpdateProjectsResult(None, False)
for project in projects:
if not self._SkipUpdatingProjectRevisionId(project):
project.SetRevisionId(commit_ids.get(project.relpath))
manifest_path = self._WriteManifestFile()
return UpdateProjectsResult(manifest_path, False)
@functools.lru_cache(maxsize=10)
def _PrintBetaNotice():
"""Print the notice of beta status."""
print(
"NOTICE: --use-superproject is in beta; report any issues to the "
"address described in `repo version`",
file=sys.stderr,
)
@functools.lru_cache(maxsize=None)
def _UseSuperprojectFromConfiguration():
"""Returns the user choice of whether to use superproject."""
user_cfg = RepoConfig.ForUser()
time_now = int(time.time())
user_value = user_cfg.GetBoolean("repo.superprojectChoice")
if user_value is not None:
user_expiration = user_cfg.GetInt("repo.superprojectChoiceExpire")
if (
user_expiration is None
or user_expiration <= 0
or user_expiration >= time_now
):
# TODO(b/190688390) - Remove prompt when we are comfortable with the
# new default value.
if user_value:
print(
(
"You are currently enrolled in Git submodules "
"experiment (go/android-submodules-quickstart). Use "
"--no-use-superproject to override.\n"
),
file=sys.stderr,
)
else:
print(
(
"You are not currently enrolled in Git submodules "
"experiment (go/android-submodules-quickstart). Use "
"--use-superproject to override.\n"
),
file=sys.stderr,
)
return user_value
# We don't have an unexpired choice, ask for one.
system_cfg = RepoConfig.ForSystem()
system_value = system_cfg.GetBoolean("repo.superprojectChoice")
if system_value:
# The system configuration is proposing that we should enable the
# use of superproject. Treat the user as enrolled for two weeks.
#
# TODO(b/190688390) - Remove prompt when we are comfortable with the new
# default value.
userchoice = True
time_choiceexpire = time_now + (86400 * 14)
user_cfg.SetString(
"repo.superprojectChoiceExpire", str(time_choiceexpire)
)
user_cfg.SetBoolean("repo.superprojectChoice", userchoice)
print(
"You are automatically enrolled in Git submodules experiment "
"(go/android-submodules-quickstart) for another two weeks.\n",
file=sys.stderr,
)
return True
# For all other cases, we would not use superproject by default.
return False
def PrintMessages(use_superproject, manifest):
"""Returns a boolean if error/warning messages are to be printed.
Args:
use_superproject: option value from optparse.
manifest: manifest to use.
"""
return use_superproject is not None or bool(manifest.superproject)
def UseSuperproject(use_superproject, manifest):
"""Returns a boolean if use-superproject option is enabled.
Args:
use_superproject: option value from optparse.
manifest: manifest to use.
Returns:
Whether the superproject should be used.
"""
if not manifest.superproject:
# This (sub) manifest does not have a superproject definition.
return False
elif use_superproject is not None:
return use_superproject
else:
client_value = manifest.manifestProject.use_superproject
if client_value is not None:
return client_value
elif manifest.superproject:
return _UseSuperprojectFromConfiguration()
else:
return False