| # Copyright 2016 The Fuchsia Authors. All rights reserved. |
| # Use of this source code is governed by a BSD-style license that can be |
| # found in the LICENSE file. |
| |
| from future.moves.urllib.parse import urlparse |
| from past.builtins import basestring |
| |
| import hashlib |
| import re |
| |
| from recipe_engine import recipe_api |
| |
| |
| class GitApi(recipe_api.RecipeApi): |
| """GitApi provides support for Git.""" |
| |
| class RebaseError(recipe_api.StepFailure): |
| pass |
| |
| _GIT_HASH_RE = re.compile("[0-9a-fA-F]{40}") |
| |
| def __init__(self, git_properties, *args, **kwargs): |
| super(GitApi, self).__init__(*args, **kwargs) |
| self._cache_path = git_properties.get("cache_path", "git") |
| |
| def __call__(self, step_name, *args, **kwargs): |
| """Return a git command step. |
| |
| Args: |
| step_name (str): Name of the step. |
| """ |
| cmd = ["git"] |
| for k, v in sorted(kwargs.pop("config", {}).items()): |
| cmd.extend(["-c", "%s=%s" % (k, v)]) |
| cmd += list(args) |
| return self.m.step(step_name, cmd, **kwargs) |
| |
| def apply(self, patch, **kwargs): |
| """Apply the specified git patch file. |
| |
| Args: |
| patch (str): Path to git patch file. |
| """ |
| return self("apply patch", "apply", patch, **kwargs) |
| |
| def branch(self, contains=None, remotes=False, step_name="git branch"): |
| """Run git branch. |
| |
| Args: |
| contains (str): Return local branches which contain this ref. |
| remotes (bool): Return only remote branches. |
| """ |
| args = ["branch"] |
| if remotes: |
| args.append("--remotes") |
| if contains: |
| args += ["--contains", contains] |
| step = self(step_name, stdout=self.m.raw_io.output_text(), *args) |
| step.presentation.logs["stdout"] = step.stdout |
| lines = step.stdout.strip().split("\n") |
| branches = [] |
| for line in lines: |
| branch = line.strip() |
| if remotes: |
| prefix = "origin/" |
| if branch.startswith(prefix): |
| branch = branch[len(prefix) :] |
| branches.append(branch) |
| return branches |
| |
| def init(self, bare=False, **kwargs): |
| """Create an empty git repository or re-initialize an existing one. |
| |
| Args: |
| bare (bool): Whether to create a bare repository. |
| """ |
| args = ["init"] |
| if bare: |
| args.append("--bare") |
| return self("git init", *args, **kwargs) |
| |
| def set_config(self, key, value=None, step_name="set config", **kwargs): |
| """Set a config. |
| |
| Args: |
| step_name (str): Name of the step. |
| key (str): Config name. |
| value (str or None): Optional config value. |
| """ |
| args = [key] |
| if value: |
| args.append(value) |
| return self(step_name, "config", *args, **kwargs) |
| |
| def config_remove_section(self, section, step_name="remove section", **kwargs): |
| """Remove section from config. |
| |
| Args: |
| step_name (str): Name of the step. |
| section (str): Section to remove. |
| """ |
| return self(step_name, "config", "--remove-section", section, **kwargs) |
| |
| def config_replace_all( |
| self, |
| name, |
| value, |
| value_regex=None, |
| step_name="git config replace all", |
| **kwargs |
| ): |
| """Replace all matching config lines. |
| |
| Args: |
| name (str): Name of config to replace. |
| value (str): Value to replace config with. |
| value_regex (str): Regex for multiline replacement. |
| """ |
| args = ["config", "--replace-all", name, value] |
| if value_regex: |
| args.append(value_regex) |
| return self(step_name, *args, **kwargs) |
| |
| def fetch( |
| self, |
| repository=None, |
| refspec=None, |
| all_remotes=False, |
| prune=False, |
| tags=False, |
| recurse_submodules=None, |
| depth=None, |
| step_name="git fetch", |
| **kwargs |
| ): |
| """Fetch a ref. |
| |
| Args: |
| repository (str): Repository to fetch. Must be set if `all_remotes` is |
| not set. |
| refspec (str): Refspec to fetch. |
| all_remotes (bool): Whether to fetch all remotes. Must be set if |
| `repository` is not set. |
| prune (bool): Whether to prune stale refs. |
| tags (bool): Whether to fetch tags. |
| recurse_submodules (bool|str|None): Whether to recurse submodules. |
| Default in Git is "on-demand" unless git config is different. |
| None means pass no options into git to use the default. |
| depth (int or None): History fetch depth. If None, do a full fetch. |
| """ |
| assert (repository and not all_remotes) or ( |
| all_remotes and not repository |
| ), "only one of `repository` or `all_remotes` can be set" |
| args = ["fetch"] |
| if all_remotes: |
| args.append("--all") |
| if prune: |
| args.append("--prune") |
| if tags: |
| args.append("--tags") |
| if repository: |
| args.append(repository) |
| if refspec: |
| args.append(refspec) |
| |
| # The default value for recurse-submodules is usually "on-demand" |
| # (unless set otherwise in git config). If not specified by the caller |
| # don't pass any --recurse-submodules argument. |
| assert recurse_submodules in (True, False, None, "on-demand") |
| if recurse_submodules == True: |
| args.append("--recurse-submodules") |
| elif recurse_submodules == False: |
| args.append("--no-recurse-submodules") |
| elif recurse_submodules == "on-demand": |
| args.append("--recurse-submodules=on-demand") # pragma: no cover |
| |
| if depth: |
| args.extend(["--depth", depth]) |
| return self(step_name, *args, **kwargs) |
| |
| def remote_add(self, remote, url, **kwargs): |
| """Add a remote. |
| |
| Args: |
| remote (str): Remote to add. |
| url (str): URL of the remote. |
| """ |
| return self("git remote", "remote", "add", remote, url, **kwargs) |
| |
| def rev_parse(self, ref, step_name="git rev-parse", **kwargs): |
| """Parse a rev into a revision. |
| |
| Args: |
| ref (str): Git ref to parse into revision. |
| """ |
| return self( |
| step_name, "rev-parse", ref, stdout=self.m.raw_io.output_text(), **kwargs |
| ).stdout.strip() |
| |
| def rev_list_count( |
| self, rev1, rev2, step_name="git rev-list --count", test_data=None |
| ): |
| """Return the number of revisions between rev1 and rev2. |
| |
| Args: |
| rev1 (str): The first git revision. |
| rev2 (str): The second git revision. |
| """ |
| num_revisions = self( |
| step_name, |
| "rev-list", |
| "--count", |
| "{rev1}..{rev2}".format(rev1=rev1, rev2=rev2), |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(test_data), |
| ).stdout.strip() |
| |
| return int(num_revisions) |
| |
| def hash_object(self, hash_input, step_name="git hash-object", step_test_data=None): |
| """Compute object ID. |
| |
| Args: |
| step_name (str): Name of the step. |
| hash_input (str): Hash input. |
| """ |
| if not step_test_data: |
| mock_hash = hashlib.sha1(hash_input.encode()).hexdigest() |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| mock_hash |
| ) |
| return self( |
| step_name, |
| "hash-object", |
| self.m.raw_io.input(hash_input), |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=step_test_data, |
| ) |
| |
| def checkout( |
| self, |
| url, |
| path=None, |
| ref=None, |
| recursive=False, |
| submodules=True, |
| submodule_force=False, |
| submodule_paths=None, |
| remote="origin", |
| checkout_file=None, |
| cache=True, |
| depth=None, |
| **kwargs |
| ): |
| """Checkout a given ref and return the checked out revision. |
| |
| Args: |
| url (str): URL of remote repo to use as upstream. |
| path (Path): Directory to clone into. |
| ref (str): Ref to fetch and checkout. |
| recursive (bool): Whether to recursively fetch submodules. |
| submodules (bool): Whether to sync and update submodules. |
| submodule_force (bool): Whether to update submodules with --force. |
| submodule_paths (list): List of path(s) to submodule(s). |
| remote (str): Name of the remote to use. |
| checkout_file (str): Optional path to a single file to checkout. |
| cache (bool): Whether to use the reference cache. |
| depth (int or None): Depth in history of checkout. If None, do a full |
| checkout. |
| """ |
| if not path: |
| path = url.rsplit("/", 1)[-1] |
| if path.endswith(".git"): # https://host/foobar.git |
| path = path[: -len(".git")] |
| path = path or path.rsplit("/", 1)[-1] # ssh://host:repo/foobar/.git |
| path = self.m.path["start_dir"].join(path) |
| |
| self.m.file.ensure_directory("makedirs", path) |
| |
| with self.m.context(cwd=path): |
| if self.m.path.exists(path.join(".git")): # pragma: no cover |
| self.config_remove_section("remote.%s" % remote, **kwargs) |
| else: |
| self.init(**kwargs) |
| self.remote_add(remote or "origin", url) |
| self.set_config( |
| "fetch.uriprotocols", |
| "https", |
| step_name="set fetch.uriprotocols", |
| **kwargs |
| ) |
| if cache: |
| with self.m.step.nest("cache"), self.m.cache.guard( |
| cache=self._cache_path |
| ): |
| o = urlparse(url) |
| dirname = o.hostname + o.path.replace("-", "--").replace("/", "-") |
| cache_path = self.m.path["cache"].join(self._cache_path, dirname) |
| |
| self.m.file.ensure_directory("makedirs", cache_path) |
| |
| with self.m.context(cwd=cache_path): |
| objects_path = cache_path.join("objects") |
| if self.m.path.exists(objects_path): # pragma: no cover |
| self.config_remove_section("remote.%s" % remote, **kwargs) |
| else: |
| self.init(bare=True, **kwargs) |
| self.set_config( |
| "remote.%s.url" % remote or "origin", |
| url, |
| step_name="remote set-url", |
| **kwargs |
| ) |
| self.set_config( |
| "fetch.uriprotocols", |
| "https", |
| step_name="set fetch.uriprotocols", |
| **kwargs |
| ) |
| self.config_replace_all( |
| "remote.origin.fetch", |
| "+refs/heads/*:refs/heads/*", |
| value_regex=r"\+refs/heads/\*:.*", |
| step_name="replace fetch configs", |
| **kwargs |
| ) |
| self.fetch( |
| repository=remote or "origin", |
| prune=True, |
| tags=True, |
| depth=depth, |
| **kwargs |
| ) |
| |
| info = path.join(".git", "objects", "info") |
| self.m.file.ensure_directory("makedirs object/info", info) |
| self.m.file.write_text( |
| "alternates", info.join("alternates"), "%s\n" % objects_path |
| ) |
| |
| if not ref: |
| fetch_ref = "main" |
| checkout_ref = "FETCH_HEAD" |
| elif self._GIT_HASH_RE.match(ref): |
| fetch_ref = "" |
| checkout_ref = ref |
| elif ref.startswith("refs/heads/"): |
| fetch_ref = ref[len("refs/heads/") :] |
| checkout_ref = "FETCH_HEAD" |
| else: |
| fetch_ref = ref |
| checkout_ref = "FETCH_HEAD" |
| self.fetch( |
| repository=remote or "origin", |
| refspec=fetch_ref, |
| tags=True, |
| recurse_submodules=recursive or None, |
| depth=depth, |
| **kwargs |
| ) |
| self.raw_checkout( |
| ref=checkout_ref, force=True, checkout_file=checkout_file, **kwargs |
| ) |
| step_test_data = kwargs.pop( |
| "step_test_data", |
| lambda: self.m.raw_io.test_api.stream_output_text("deadbeef"), |
| ) |
| sha = self.rev_parse("HEAD", step_test_data=step_test_data) |
| self.clean(force=True, recursive=True, ignore_rules=True, **kwargs) |
| if submodules: |
| with self.m.step.nest("submodule"): |
| self.sync_submodule(**kwargs) |
| self.update_submodule( |
| recursive=recursive, |
| force=submodule_force, |
| paths=submodule_paths, |
| **kwargs |
| ) |
| return sha |
| |
| def clean( |
| self, |
| step_name="git clean", |
| force=False, |
| recursive=False, |
| ignore_rules=False, |
| **kwargs |
| ): |
| """Remove untracked files from working tree. |
| |
| Args: |
| force (bool): Clean with force. |
| recursive (bool): Recurse into directories. |
| ignore_rules (bool): -x option: use ignore rules from -e. |
| """ |
| args = ["clean"] |
| if force: |
| args.append("-f") |
| if recursive: |
| args.append("-d") |
| if ignore_rules: |
| args.append("-x") |
| return self(step_name, *args, **kwargs) |
| |
| def raw_checkout( |
| self, |
| step_name="git checkout", |
| ref=None, |
| checkout_file=None, |
| force=False, |
| branch=None, |
| directory=None, |
| **kwargs |
| ): |
| """Run a git checkout. |
| |
| Args: |
| step_name (str): Name of the step. |
| ref (str or None): Ref to checkout. |
| checkout_file (str or None): Optional path to a single file to |
| checkout. |
| force (bool): Checkout with force. |
| branch (str): Checkout new branch with this name. |
| directory (Path): Checkout in this directory. TODO(atyfto): Move |
| this to __call__. |
| """ |
| args = [] |
| if directory: |
| args += ["-C", directory] |
| args.append("checkout") |
| if force: |
| args.append("-f") |
| if branch: |
| args += ["-b", branch] |
| if ref: |
| args.append(ref) |
| if checkout_file: |
| args += ["--", checkout_file] |
| return self(step_name, *args, **kwargs) |
| |
| def sync_submodule(self, step_name="git submodule sync", **kwargs): |
| """Sync submodule. |
| |
| Args: |
| step_name (str): Name of the step. |
| """ |
| return self(step_name, "submodule", "sync", **kwargs) |
| |
| def update_submodule( |
| self, |
| step_name="git submodule update", |
| recursive=False, |
| force=False, |
| paths=None, |
| **kwargs |
| ): |
| """Update submodule. |
| |
| Args: |
| step_name (str): Name of the step. |
| recursive (bool): Recursively update submodule. |
| force (bool): Force-update submodule. |
| submodule_path (str): Path(s) to submodule(s). |
| """ |
| args = ["submodule", "update", "--init"] |
| if recursive: |
| args.append("--recursive") |
| if force: |
| args.append("--force") |
| args.extend(("--jobs", str(min(16, max(2, self.m.platform.cpu_count // 2))))) |
| if paths: |
| args.extend(paths) |
| return self(step_name, *args, **kwargs) |
| |
| def checkout_commit(self, commit, path, **kwargs): |
| """Checkout a gitiles commit. |
| |
| Args: |
| (see `checkout()` for others) |
| commit (common_pb2.GitilesCommit): The commit to check out. |
| """ |
| url = "https://%s/%s" % (commit.host, commit.project) |
| return self.checkout(url, path=path, ref=commit.id, **kwargs) |
| |
| def checkout_cl( |
| self, cl, path, rebase_merges=False, onto="refs/heads/main", **kwargs |
| ): |
| """Checkout a CL and rebase it onto ref. |
| |
| Args: |
| (see `checkout()` for others) |
| cl (common_pb2.GerritChange or None): The CL to check out. |
| onto (str): The ref to rebase the CL onto. |
| """ |
| url = "https://%s/%s" % (cl.host.replace("-review", ""), cl.project) |
| ref = "refs/changes/%02d/%d/%d" % (cl.change % 100, cl.change, cl.patchset) |
| with self.m.step.nest("fetch %s" % onto): |
| onto = self.checkout(url, path=path, ref=onto, **kwargs) |
| checkout_result = self.checkout(url, path=path, ref=ref, **kwargs) |
| with self.m.context(cwd=path): |
| self.rebase(onto, rebase_merges=rebase_merges) |
| return checkout_result |
| |
| def checkout_from_build_input( |
| self, path, repo, rebase_merges=False, fallback_ref=None, **kwargs |
| ): |
| """Use the Buildbucket input to checkout the triggering revision. |
| |
| Checkout the triggering CL in CQ, or the triggering commit in CI. |
| If repo does not match the build input, fallback to checkout the repo |
| at HEAD, or the fallback ref if specified. |
| """ |
| with self.m.step.nest("checkout"), self.m.context(infra_steps=True): |
| build_input = self.m.buildbucket.build.input |
| if build_input.gerrit_changes: |
| change = build_input.gerrit_changes[0] |
| if ( |
| "https://%s/%s" |
| % (change.host.replace("-review", ""), change.project) |
| == repo |
| ): |
| return self.checkout_cl( |
| change, path=path, rebase_merges=rebase_merges, **kwargs |
| ) |
| elif build_input.gitiles_commit.project: |
| gitiles_commit = build_input.gitiles_commit |
| if ( |
| "https://%s/%s" % (gitiles_commit.host, gitiles_commit.project) |
| == repo |
| ): |
| return self.checkout_commit( |
| build_input.gitiles_commit, path=path, **kwargs |
| ) |
| return self.checkout(repo, path=path, ref=fallback_ref, **kwargs) |
| |
| def commit( |
| self, |
| message=None, |
| no_edit=False, |
| step_name="git commit", |
| files=(), |
| amend=False, |
| allow_empty=False, |
| all_tracked=False, |
| all_files=False, |
| author_override=None, |
| **kwargs |
| ): |
| """Run git commit in the current working directory. |
| |
| Args: |
| message (str): The message to attach to the commit. Mutually exclusive |
| with `no_edit`. |
| no_edit (bool): Do not edit commit message. Mutually exclusive with |
| `message`. |
| files (seq[Path]): The set of files containing changes to commit. |
| amend (bool): Whether to amend the previous commit. |
| allow_empty (bool): Whether to allow an empty commit. |
| all_tracked (bool): Stage all tracked files before committing. If True, |
| files must be empty and all_files must be False. |
| all_files (bool): Stage all files (even untracked) before committing. If |
| True, files must be empty and all_tracked must be False. |
| author_override (dict): Dict representation of a git.Commit_User |
| containing name and email keys. If specified, override the author of |
| the commit. |
| """ |
| assert (message and not no_edit) or (not message and no_edit) |
| cmd = ["commit"] |
| if message: |
| cmd += ["-m", message] |
| if no_edit: |
| cmd.append("--no-edit") |
| if amend: |
| cmd.append("--amend") |
| if allow_empty: |
| cmd.append("--allow-empty") |
| if all_tracked: |
| assert not all_files |
| assert not files |
| cmd.append("-a") |
| elif all_files: |
| assert not files |
| self.add(add_all=True) |
| else: |
| cmd.extend(files) |
| if author_override: |
| cmd += [ |
| "--author", |
| "%s <%s>" % (author_override["name"], author_override["email"]), |
| ] |
| return self(step_name, *cmd, **kwargs) |
| |
| def tag(self, name, step_name="git tag"): |
| """Create a tag. |
| |
| Args: |
| name (str): Name of tag. |
| """ |
| return self(step_name, "tag", name) |
| |
| def add( |
| self, |
| step_name="git add", |
| pathspec=None, |
| add_all=False, |
| intent_to_add=False, |
| only_tracked=False, |
| force=False, |
| **kwargs |
| ): |
| """Add file contents to index. |
| |
| Args: |
| step_name (str): Name of the step. |
| pathspec (str): Add pathspec to index. |
| add_all (bool): Make index match the working tree. |
| intent_to_add (bool): Move any new files into tracked state. |
| only_tracked (bool): Add all tracked files to the index, and |
| no untracked files implies add_all=False. |
| force (bool): Allow adding otherwise ignored files. |
| """ |
| args = ["add"] |
| if only_tracked: |
| assert not add_all, "only_tracked and add_all cannot be set together" |
| args.append("--update") |
| if add_all: |
| args.append("--all") |
| if intent_to_add: |
| args.append("--intent-to-add") |
| if force: |
| args.append("--force") |
| if pathspec: |
| args.append(pathspec) |
| return self(step_name, *args, **kwargs) |
| |
| def ls_files( |
| self, |
| step_name="git ls-files", |
| modified=False, |
| deleted=False, |
| others=False, |
| exclude_standard=False, |
| line_termination=False, |
| file=None, |
| test_data="hello", |
| **kwargs |
| ): |
| """ |
| List files in index. |
| |
| Args: |
| step_name (str): Name of the step. |
| modified (bool): Include modified files. |
| deleted (bool): Include deleted files. |
| others (bool): Include untracked files. |
| exclude_standard (bool): Add standard Git exclusions. |
| line_termination (bool): -z option: null byte termination on output. |
| file (str): Filepath or regex to filter for. |
| test_data (str): Test data stdout. |
| """ |
| args = ["ls-files"] |
| if modified: |
| args.append("--modified") |
| if deleted: |
| args.append("--deleted") |
| if exclude_standard: |
| args.append("--exclude-standard") |
| if others: |
| args.append("--others") |
| if line_termination: |
| args.append("-z") |
| if file: |
| args += ["--", file] |
| return self( |
| step_name, |
| *args, |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=lambda: self.m.raw_io.test_api.stream_output_text(test_data), |
| **kwargs |
| ) |
| |
| def diff( |
| self, |
| step_name="git diff", |
| ref_base="HEAD", |
| ref=None, |
| patch_file=None, |
| unified=None, |
| cached=False, |
| exit_code=False, |
| step_test_data=None, |
| **kwargs |
| ): |
| """Run git diff in the current working directory. |
| |
| Args: |
| step_name (str): Name of the step. |
| ref_base (str or None): Base ref for diff. |
| ref (str or None): Secondary ref for diff. |
| patch_file (str): Path to patch file. |
| unified (int): Generate diff with `unified` lines of context. |
| cached (bool): Get staged files. |
| exit_code (bool): Exit with codes similar to diff(1). |
| """ |
| if step_test_data is None: |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| """ |
| diff --git a/recipe_modules/fuchsia/api.py b/recipe_modules/fuchsia/api.py |
| index 429e4612..83925ff3 100644 |
| --- a/recipe_modules/fuchsia/api.py |
| +++ b/recipe_modules/fuchsia/api.py |
| @@ -562,15 +562,6 @@ class FuchsiaApi(recipe_api.RecipeApi): |
| - # arm64 hardware targets have a stable version of Zedboot on their |
| - # recovery partition, we do not want to send the build Zedboot to |
| - # overwrite it. |
| - if build.target == 'arm64' and device_type != 'QEMU': |
| - for image in build.images.keys(): |
| - # Matches both 'zircon-r' and 'zircon-r.signed' |
| - if 'zircon-r' in image: |
| - del build.images[image] |
| """ |
| ) |
| args = ["diff"] |
| if ref_base: |
| args.append(ref_base) |
| if ref: |
| args.append(ref) |
| if unified is not None: |
| args.append("--unified=%d" % unified) |
| if cached: |
| args.append("--cached") |
| if exit_code: |
| args.append("--exit-code") |
| return self( |
| step_name, |
| *args, |
| step_test_data=step_test_data, |
| stdout=self.m.raw_io.output_text(leak_to=patch_file), |
| **kwargs |
| ) |
| |
| def log(self, step_name="git log", directory=None, depth=10): |
| """Get git log. |
| |
| Args: |
| step_name (str): Name of the step. |
| directory (Path or None): Directory to run git log in. Defaults to |
| cwd. TODO(atyfto): Move this to __call__. |
| depth (int): Commit depth of log. |
| """ |
| args = [] |
| if directory: |
| args += ["-C", directory] |
| args += ["log", "-n", depth] |
| return self(step_name, *args) |
| |
| def push( |
| self, |
| refs, |
| step_name="git push", |
| remote="origin", |
| atomic=False, |
| options=(), |
| dryrun=False, |
| **kwargs |
| ): |
| """Run a git push for a ref or sequence of refs. |
| |
| Args: |
| refs (str or seq(str)): Ref or sequence of refs to push to remote. |
| remote (str): Remote repository to push to. |
| atomic (bool): Request atomic transaction on remote side. |
| NOTE: This option is ignored by Gerrit. |
| options (seq(str)): Server-specific options passed to the --push-option |
| flag. |
| dryrun (bool): Push in dryrun mode. |
| """ |
| args = ["push"] |
| if atomic: |
| args.append("--atomic") |
| if dryrun: |
| args.append("--dry-run") |
| for option in options: |
| args.extend(["--push-option", option]) |
| args.append(remote) |
| if isinstance(refs, basestring): |
| args.append(refs) |
| else: |
| args.extend(refs) |
| return self(step_name, *args, **kwargs) |
| |
| def cherry_pick(self, commit, strategy_option=None, step_name="git cherry-pick"): |
| """Cherry-pick a commit. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Commit to cherry-pick as sha1. |
| strategy (str|None): Use merge strategy option. |
| """ |
| args = ["cherry-pick", commit] |
| if strategy_option: |
| args += ["--strategy-option", strategy_option] |
| return self(step_name, *args) |
| |
| def revert(self, commit, strategy_option=None, step_name="git revert"): |
| """Revert a commit. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Commit to revert as sha1. |
| strategy (str|None): Use merge strategy option. |
| """ |
| args = ["revert", commit] |
| if strategy_option: |
| args += ["--strategy-option", strategy_option] |
| return self(step_name, *args) |
| |
| def rebase(self, ref, rebase_merges=False, **kwargs): |
| """Run rebase with the given arguments. |
| |
| If a StepFailure occurs, abort the rebase and re-raise the |
| error. |
| """ |
| args = ["rebase"] |
| if rebase_merges: |
| args.append("--rebase-merges") |
| args.append(ref) |
| try: |
| self("git rebase", *args, **kwargs) |
| except self.m.step.StepFailure: # pragma: no cover |
| self("git rebase abort", "rebase", "--abort", **kwargs) |
| raise self.RebaseError("Failed to rebase") |
| |
| def get_hash(self, commit="HEAD", **kwargs): |
| """Find and return the hash of the given commit.""" |
| step_test_data = kwargs.pop( |
| "step_test_data", |
| lambda: self.m.raw_io.test_api.stream_output_text("deadbeef"), |
| ) |
| return self.rev_parse(commit, step_test_data=step_test_data) |
| |
| def describe( |
| self, |
| step_name="git describe", |
| commit="HEAD", |
| tags=True, |
| contains=False, |
| exact_match=True, |
| expected_num=None, |
| test_data=None, |
| **kwargs |
| ): |
| """Describe a commit by its tag(s). |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Ref to describe. |
| tags (bool): If True, use any tag in refs/tags namespace. Otherwise, |
| only use annotated tags. |
| contains (bool): If True, find the tag that comes after the given commit. |
| Automatically implies 'tags' flag. |
| exact_match (bool): If True, only output exact tag matches to the given |
| commit. |
| expected_num (int): If specified, raise exception if expected number |
| of tags which describe the given commit is not equal to actual. |
| test_data (func): If specified, overwrite step test data. |
| |
| Returns: |
| list: tag(s) describing the commit. |
| """ |
| if test_data is None: |
| test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| "testtag1\ntesttag2" |
| ) |
| cmd = ["describe"] |
| if tags is True: |
| cmd.append("--tags") |
| if contains is True: |
| cmd.append("--contains") |
| if exact_match is True: |
| cmd.append("--exact-match") |
| cmd.append(commit) |
| step = self( |
| step_name, |
| *cmd, |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=test_data, |
| ok_ret="any", |
| **kwargs |
| ) |
| # An error will be raised if the commit has no tags to describe it. |
| tags = [] if step.retcode else step.stdout.strip().split("\n") |
| if expected_num is not None and len(tags) != expected_num: |
| raise self.m.step.StepFailure( |
| "Expected number of matching tags ({expected_num}) does not match " |
| "actual ({actual_num}).".format( |
| expected_num=expected_num, |
| actual_num=len(tags), |
| ) |
| ) |
| return tags |
| |
| def get_remote_url(self, name, step_test_data=None, **kwargs): |
| """Find and return the URL of given remote.""" |
| if not step_test_data: |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| "https://fuchsia.googlesource.com/fuchsia" |
| ) |
| return self.set_config( |
| "remote.%s.url" % name, |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=step_test_data, |
| **kwargs |
| ).stdout.strip() |
| |
| def get_timestamp(self, commit="HEAD", step_name="get commit ts", **kwargs): |
| """Find and return the timestamp of the given commit. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Ref to get timestamp for. |
| """ |
| step_test_data = kwargs.pop( |
| "step_test_data", |
| lambda: self.m.raw_io.test_api.stream_output_text("1473312770"), |
| ) |
| args = ["show", commit, "--format=%at", "--no-patch"] |
| return self( |
| step_name, |
| *args, |
| step_test_data=step_test_data, |
| stdout=self.m.raw_io.output_text(), |
| **kwargs |
| ).stdout.strip() |
| |
| def get_commit_message( |
| self, step_name="get commit msg", commit="HEAD", oneline=False, **kwargs |
| ): |
| """Get message of the given commit. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Git ref to get message for. |
| oneline (bool): Whether to shorten message to one line. |
| """ |
| step_test_data = kwargs.pop( |
| "step_test_data", |
| lambda: self.m.raw_io.test_api.stream_output_text("[foo] bar\nbaz\n"), |
| ) |
| args = [ |
| "show", |
| commit, |
| "--format=%B", |
| "--no-patch", |
| ] |
| if oneline is True: |
| args.append("--oneline") |
| step = self( |
| step_name, |
| *args, |
| step_test_data=step_test_data, |
| stdout=self.m.raw_io.output_text(), |
| **kwargs |
| ) |
| commit_msg = step.stdout.strip() |
| step.presentation.logs["commit message"] = commit_msg.splitlines() |
| return commit_msg |
| |
| def get_changed_files( |
| self, |
| step_name="git diff-tree", |
| commit="HEAD", |
| deleted=True, |
| ignore_submodules=False, |
| test_data=("foo/bar/baz", "foo/baz"), |
| **kwargs |
| ): |
| """Return a list of files changed/added/removed in a CL. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit (str): Git ref to get the files for. |
| deleted (bool): Whether to include deleted files. |
| ignore_submodules (bool): Whether to include submodule paths. If |
| True, the list may contain paths to submodule directories, but |
| not paths to changed files within submodules. |
| test_data (seq of str): Mock list of changed files. |
| **kwargs (dict): Passed through to __call__(). |
| """ |
| args = [ |
| "diff-tree", |
| "--no-commit-id", |
| "--name-only", |
| "-r", # Recurse into subrepos. |
| # Separate results with nuls - otherwise git will do really ugly |
| # escaping of non-ASCII characters in file names. |
| "-z", |
| ] |
| if not deleted: |
| args.append("--diff-filter=d") |
| if ignore_submodules: |
| args.append("--ignore-submodules=all") |
| args.append(commit) |
| |
| step = self( |
| step_name, |
| *args, |
| step_test_data=lambda: self.m.raw_io.test_api.stream_output_text( |
| "".join("{}\0".format(f) for f in test_data) |
| ), |
| stdout=self.m.raw_io.output_text(), |
| **kwargs |
| ) |
| files = step.stdout.strip("\0").split("\0") |
| step.presentation.logs["files"] = files |
| return files |
| |
| def snap_branch( |
| self, |
| url, |
| snap_ref, |
| branch, |
| message, |
| tmp_branch="tmp_branch", |
| path=None, |
| checkout=True, |
| push=False, |
| step_name="snap branch", |
| ): |
| """Snap a target branch to specified ref, i.e. overwrite target branch |
| to be equal to specified ref without losing previous history. |
| |
| Args: |
| url (str): url of remote repo to use as upstream. |
| snap_ref (str): commit hash ref to snap to. |
| branch (str): target branch to apply snap. |
| message (str): snap commit message. |
| tmp_branch (str): temporary branch name. |
| path (Path): directory to checkout in. |
| checkout (bool): if False, use the existing checkout at path. |
| push (bool): push snap to target branch. |
| """ |
| assert not ( |
| not checkout and not path |
| ), "Cannot attempt snap with neither a checkout nor a path." |
| with self.m.step.nest(step_name): |
| path = path or self.m.path.mkdtemp("snap_branch") |
| with self.m.context(cwd=path): |
| if checkout: |
| self.checkout(url=url, ref=branch, path=path) |
| # If branch HEAD is already at snap ref, no further action needed. |
| if self.get_hash() == snap_ref: |
| return |
| |
| remote_branch = "origin/{branch}".format(branch=branch) |
| # Create temporary branch off target branch. |
| self.raw_checkout(ref=remote_branch, branch=tmp_branch) |
| # Reset temporary branch to snap ref, preserving history. |
| self.reset(snap_ref, hard=True) |
| self.merge(remote_branch, strategy="ours", message=message) |
| if push: |
| # Push snapped temporary branch to target branch. |
| self.push( |
| "{tmp_branch}:refs/heads/{branch}".format( |
| tmp_branch=tmp_branch, |
| branch=branch, |
| ) |
| ) |
| |
| def merge(self, ref, step_name="git merge", strategy=None, message=None): |
| """Merge ref into current branch. |
| |
| Args: |
| step_name (str): Name of the step. |
| strategy (str or None): Merge strategy. See git-merge docs. |
| message (str or None): Commit message for merge commit in case one |
| is created. |
| """ |
| args = ["merge"] |
| if strategy: |
| args += ["--strategy", strategy] |
| args.append(ref) |
| if message: |
| args += ["--message", message] |
| self(step_name, *args) |
| |
| def is_ancestor(self, step_name="is ancestor", commit_1="HEAD", commit_2="HEAD"): |
| """Check if commit_1 is an ancestor of commit_2. |
| |
| Args: |
| step_name (str): Name of the step. |
| commit_1 (str): A commit that should be an ancestor of commit_2. |
| commit_2 (str): A commit we are checking ancestor against. |
| |
| Returns: |
| bool: True if commit_1 is ancestor of commit_2, otherwise False. |
| |
| Raises: |
| StepFailure: git command returns exit code >=2. |
| """ |
| args = ["merge-base", "--is-ancestor", commit_1, commit_2] |
| return self(step_name, *args, ok_ret=(0, 1)).retcode == 0 |
| |
| def reset(self, ref, step_name="git reset", hard=False): |
| """Reset current HEAD to specified ref. |
| |
| Args: |
| step_name (str): Name of the step. |
| hard (bool): Reset index and working tree. |
| """ |
| args = ["reset"] |
| if hard: |
| args.append("--hard") |
| args.append(ref) |
| self(step_name, *args) |
| |
| def get_default_remote_branch( |
| self, url, step_name="git ls-remote --symref HEAD", step_test_data=None |
| ): |
| """Get default remote branch. |
| |
| Args: |
| step_name (str): Name of the step. |
| url (str): remote repository URL. |
| |
| Returns: |
| str: Full name of ref, often including "refs/heads/". |
| """ |
| if not step_test_data: |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| "ref: refs/heads/main HEAD\nh3ll0 HEAD" |
| ) |
| step = self( |
| step_name, |
| "ls-remote", |
| "--symref", |
| url, |
| "HEAD", |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=step_test_data, |
| ) |
| for line in step.stdout.splitlines(): |
| line = line.strip() |
| if line.startswith("ref:"): |
| return line.split()[1] |
| raise ValueError( # pragma: no cover |
| "expected 'ref:' line in {!r}".format(step.stdout) |
| ) |
| |
| def get_all_remote_branch_heads( |
| self, url, branch=None, step_name="git ls-remote", step_test_data=None |
| ): |
| """Get remote branch head(s). |
| |
| If branch is specified, the number of entries returned will be either |
| one or zero, depending if a branch with the name `branch` is found. |
| |
| Args: |
| step_name (str): Name of the step. |
| url (str): remote repository URL. |
| branch (str or None): The branch name to pass into `git ls-remote`. |
| |
| Returns: |
| dict: A mapping between branche(s) and their HEAD commit SHA1s. |
| """ |
| if not step_test_data: |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| "h3ll0\trefs/heads/{branch}".format(branch=branch) |
| if branch |
| else "h3ll0\trefs/heads/main\nh3ll01\trefs/heads/custom_branch" |
| ) |
| args = ["ls-remote", "--heads", url] + ([branch] if branch else []) |
| step = self( |
| step_name, |
| *args, |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=step_test_data |
| ) |
| stdout = step.stdout.strip() |
| heads = [line.split() for line in stdout.split("\n")] if stdout else [] |
| # Build a dictionary of names to commit SHA1s. |
| return {name: rev for rev, name in heads} |
| |
| def get_remote_branch_head( |
| self, url, branch, step_name="git ls-remote", step_test_data=None |
| ): |
| """Get remote branch head. |
| |
| If the branch does not exist remotely, return empty string. |
| |
| Args: |
| step_name (str): Name of the step. |
| url (str): remote repository URL. |
| branch (str): branch name. |
| |
| Returns: |
| str: HEAD commit SHA1 if it exists, otherwise empty string. |
| """ |
| # Query branch name to head. |
| heads = self.get_all_remote_branch_heads( |
| url, branch=branch, step_name=step_name, step_test_data=step_test_data |
| ) |
| |
| # Get branch hash or empty string if it doesn't exist. This assumes that |
| # heads has a maxiumum of 1 entry since `branch` is passed in. |
| return next(iter(heads.values()), "") |
| |
| def get_remote_tag(self, url, tag, step_name="get remote tag", step_test_data=None): |
| """Get remote tag commit. |
| |
| If the tag does not exist remotely, return empty string. |
| |
| Args: |
| step_name (str): Name of the step. |
| url (str): Remote repository URL. |
| tag (str): Git tag. |
| |
| Returns: |
| str: tag's commit SHA1 if it exists, otherwise empty string. |
| """ |
| if not step_test_data: |
| step_test_data = lambda: self.m.raw_io.test_api.stream_output_text( |
| "tagcommit {tag}".format(tag=tag) |
| ) |
| tag_output = self( |
| step_name, |
| "ls-remote", |
| "--tags", |
| url, |
| tag, |
| stdout=self.m.raw_io.output_text(), |
| step_test_data=step_test_data, |
| ).stdout.strip() |
| # Parse out the commit SHA1. |
| return tag_output.split()[0] if tag_output else tag_output |
| |
| def read_commit_footer(self, commit_msg, footer, sep=": "): |
| """Given a commit message, parse a footer value from it. |
| |
| Args: |
| commit_msg (str): The commit message to parse. |
| footer (str): The name of the footer to parse, e.g. "Change-Id". |
| sep (str or seq(str)): The separator between footer name and footer value. |
| If multiple separators are allowed (e.g. "=" and ": "), they |
| can be passed as a list. |
| |
| Returns (str): The footer value if the footer was found in the commit |
| message, otherwise None. |
| """ |
| if isinstance(sep, str): |
| sep = [sep] |
| possible_prefixes = [footer + s for s in sep] |
| for line in reversed(commit_msg.splitlines()): |
| for prefix in possible_prefixes: |
| if line.startswith(prefix): |
| return line[len(prefix) :].strip() |
| return None |
