• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#
2# Copyright (C) 2023 The Android Open Source Project
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8#      http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15#
16"""APIs for interacting with git repositories."""
17# TODO: This should be partially merged with the git_utils APIs.
18# The bulk of this should be lifted out of the tests and used by the rest of
19# external_updater, but we'll want to keep a few of the APIs just in the tests because
20# they're not particularly sensible elsewhere (specifically the shorthand for commit
21# with the update_files and delete_files arguments). It's probably easiest to do that by
22# reworking the git_utils APIs into a class like this and then deriving this one from
23# that.
24from __future__ import annotations
25
26import subprocess
27from pathlib import Path
28
29
30class GitRepo:
31    """A git repository for use in tests."""
32
33    def __init__(self, path: Path) -> None:
34        self.path = path
35
36    def run(self, command: list[str]) -> str:
37        """Runs the given git command in the repository, returning the output."""
38        return subprocess.run(
39            ["git", "-C", str(self.path)] + command,
40            check=True,
41            capture_output=True,
42            text=True,
43        ).stdout
44
45    def init(self, branch_name: str | None = None) -> None:
46        """Initializes a new git repository."""
47        self.path.mkdir(parents=True)
48        cmd = ["init"]
49        if branch_name is not None:
50            cmd.extend(["-b", branch_name])
51        self.run(cmd)
52
53    def head(self) -> str:
54        """Returns the SHA of the current HEAD."""
55        return self.run(["rev-parse", "HEAD"]).strip()
56
57    def sha_of_ref(self, ref: str) -> str:
58        """Returns the sha of the given ref."""
59        return self.run(["rev-list", "-n", "1", ref]).strip()
60
61    def current_branch(self) -> str:
62        """Returns the name of the current branch."""
63        return self.run(["branch", "--show-current"]).strip()
64
65    def fetch(self, ref_or_repo: str | GitRepo) -> None:
66        """Fetches the given ref or repo."""
67        if isinstance(ref_or_repo, GitRepo):
68            ref_or_repo = str(ref_or_repo.path)
69        self.run(["fetch", ref_or_repo])
70
71    def commit(
72        self,
73        message: str,
74        allow_empty: bool = False,
75        update_files: dict[str, str] | None = None,
76        delete_files: set[str] | None = None,
77    ) -> None:
78        """Create a commit in the repository."""
79        if update_files is None:
80            update_files = {}
81        if delete_files is None:
82            delete_files = set()
83
84        for delete_file in delete_files:
85            self.run(["rm", delete_file])
86
87        for update_file, contents in update_files.items():
88            (self.path / update_file).write_text(contents, encoding="utf-8")
89            self.run(["add", update_file])
90
91        commit_cmd = ["commit", "-m", message]
92        if allow_empty:
93            commit_cmd.append("--allow-empty")
94        self.run(commit_cmd)
95
96    def merge(
97        self,
98        ref: str,
99        allow_fast_forward: bool = True,
100        allow_unrelated_histories: bool = False,
101    ) -> None:
102        """Merges the upstream ref into the repo."""
103        cmd = ["merge"]
104        if not allow_fast_forward:
105            cmd.append("--no-ff")
106        if allow_unrelated_histories:
107            cmd.append("--allow-unrelated-histories")
108        self.run(cmd + [ref])
109
110    def switch_to_new_branch(self, name: str, start_point: str | None = None) -> None:
111        """Creates and switches to a new branch."""
112        args = ["switch", "--create", name]
113        if start_point is not None:
114            args.append(start_point)
115        self.run(args)
116
117    def tag(self, name: str, ref: str | None = None) -> None:
118        """Creates a tag at the given ref, or HEAD if not provided."""
119        args = ["tag", name]
120        if ref is not None:
121            args.append(ref)
122        self.run(args)
123
124    def commit_message_at_revision(self, revision: str) -> str:
125        """Returns the commit message of the given revision."""
126        # %B is the raw commit body
127        # %- eats the separator newline
128        # Note that commit messages created with `git commit` will always end with a
129        # trailing newline.
130        return self.run(["log", "--format=%B%-", "-n1", revision])
131
132    def file_contents_at_revision(self, revision: str, path: str) -> str:
133        """Returns the commit message of the given revision."""
134        # %B is the raw commit body
135        # %- eats the separator newline
136        return self.run(["show", "--format=%B%-", f"{revision}:{path}"])
137