def _rebuild_git()

in src/buildstream_plugins/sources/git.py [0:0]

• 106 lines of code
• 10 McCabe index (conditional complexity)

    def _rebuild_git(self, fullpath):
        if not self.tags:
            return

        with self.source.tempdir() as tmpdir:
            included = set()
            shallow = set()
            for _, commit_ref, _ in self.tags:

                if commit_ref == self.ref:
                    # rev-list does not work in case of same rev
                    shallow.add(self.ref)
                else:
                    _, out = self.source.check_output(
                        [
                            self.source.host_git,
                            "rev-list",
                            "--ancestry-path",
                            "--boundary",
                            "{}..{}".format(commit_ref, self.ref),
                        ],
                        fail="Failed to get git history {}..{} in directory: {}".format(
                            commit_ref, self.ref, fullpath
                        ),
                        fail_temporarily=True,
                        cwd=self.mirror,
                    )
                    self.source.warn("refs {}..{}: {}".format(commit_ref, self.ref, out.splitlines()))
                    for line in out.splitlines():
                        rev = line.lstrip("-")
                        if line[0] == "-":
                            shallow.add(rev)
                        else:
                            included.add(rev)

            shallow -= included
            included |= shallow

            self.source.call(
                [self.source.host_git, "init"],
                fail="Cannot initialize git repository: {}".format(fullpath),
                cwd=fullpath,
            )

            for rev in included:
                with TemporaryFile(dir=tmpdir) as commit_file:
                    self.source.call(
                        [self.source.host_git, "cat-file", "commit", rev],
                        stdout=commit_file,
                        fail="Failed to get commit {}".format(rev),
                        cwd=self.mirror,
                    )
                    commit_file.seek(0, 0)
                    self.source.call(
                        [
                            self.source.host_git,
                            "hash-object",
                            "-w",
                            "-t",
                            "commit",
                            "--stdin",
                        ],
                        stdin=commit_file,
                        fail="Failed to add commit object {}".format(rev),
                        cwd=fullpath,
                    )

            with open(
                os.path.join(fullpath, ".git", "shallow"),
                "w",
                encoding="utf-8",
            ) as shallow_file:
                for rev in shallow:
                    shallow_file.write("{}\n".format(rev))

            for tag, commit_ref, annotated in self.tags:
                if annotated:
                    with TemporaryFile(dir=tmpdir) as tag_file:
                        tag_data = "object {}\ntype commit\ntag {}\ntagger Unspecified Tagger <unspecified-tagger> 0 +0000\n".format(
                            commit_ref, tag
                        )
                        tag_file.write(tag_data.encode("ascii"))
                        tag_file.seek(0, 0)
                        _, tag_ref = self.source.check_output(
                            [
                                self.source.host_git,
                                "hash-object",
                                "-w",
                                "-t",
                                "tag",
                                "--stdin",
                            ],
                            stdin=tag_file,
                            fail="Failed to add tag object {}".format(tag),
                            cwd=fullpath,
                        )

                    self.source.call(
                        [self.source.host_git, "tag", tag, tag_ref.strip()],
                        fail="Failed to tag: {}".format(tag),
                        cwd=fullpath,
                    )
                else:
                    self.source.call(
                        [self.source.host_git, "tag", tag, commit_ref],
                        fail="Failed to tag: {}".format(tag),
                        cwd=fullpath,
                    )

            with open(os.path.join(fullpath, ".git", "HEAD"), "w", encoding="utf-8") as head:
                self.source.call(
                    [self.source.host_git, "rev-parse", self.ref],
                    stdout=head,
                    fail="Failed to parse commit {}".format(self.ref),
                    cwd=self.mirror,
                )