def apply_build()

in libmozevent/mercurial.py [0:0]

• 74 lines of code
• 15 McCabe index (conditional complexity)

    def apply_build(self, build):
        """
        Apply a stack of patches to mercurial repo
        and commit them one by one
        """
        assert isinstance(build, PhabricatorBuild)
        assert len(build.stack) > 0, "No patches to apply"
        assert all(map(lambda p: isinstance(p, PhabricatorPatch), build.stack))

        # Find the first unknown base revision
        needed_stack = []
        for patch in reversed(build.stack):
            needed_stack.insert(0, patch)

            # Stop as soon as a base revision is available
            if self.has_revision(patch.base_revision):
                logger.info("Stopping at revision {}".format(patch.base_revision))
                break

        if not needed_stack:
            logger.info("All the patches are already applied")
            return

        hg_base = self.get_base_identifier(needed_stack)

        # When base revision is missing, update to default revision
        build.base_revision = hg_base
        # TODO: Re-enable base revision identification after https://github.com/mozilla/libmozevent/issues/110.
        build.missing_base_revision = False  # not self.has_revision(hg_base)
        if build.missing_base_revision:
            logger.warning(
                "Missing base revision from Phabricator",
                revision=hg_base,
                fallback=self.default_revision,
            )
            hg_base = self.default_revision
        hg_base = self.default_revision

        # Store the actual base revision we used
        build.actual_base_revision = hg_base

        # Update the repo to base revision
        try:
            logger.info("Updating repo to revision {}".format(hg_base))
            self.repo.update(rev=hg_base, clean=True)

            # See if the repo is clean
            repo_status = self.repo.status(
                modified=True, added=True, removed=True, deleted=True
            )
            if len(repo_status) != 0:
                logger.warn(
                    "Repo is dirty! Let's clean it first.",
                    revision=hg_base,
                    repo=self.name,
                    repo_status=repo_status,
                )
                # Clean the repo - This is a workaround for Bug 1720302
                self.repo.update(rev="null", clean=True)
                # Redo the update to the correct revision
                self.repo.update(rev=hg_base, clean=True)

        except hglib.error.CommandError:
            raise Exception("Failed to update to revision {}".format(hg_base))

        # In this case revision is `hg_base`
        logger.info("Updated repo", revision=hg_base, repo=self.name)

        def get_author(commit):
            """Helper to build a mercurial author from Phabricator data"""
            author = commit.get("author")
            if author is None:
                return DEFAULT_AUTHOR
            if author["name"] and author["email"]:
                # Build clean version without quotes
                return f"{author['name']} <{author['email']}>"
            return author["raw"]

        for patch in needed_stack:
            if patch.commits:
                # Use the first commit only
                commit = patch.commits[0]
                message = "{}\n".format(commit["message"])
                user = get_author(commit)
            else:
                # We should always have some commits here
                logger.warning("Missing commit on patch", id=patch.id)
                message = ""
                user = DEFAULT_AUTHOR
            message += "Differential Diff: {}".format(patch.phid)

            logger.info("Applying patch", phid=patch.phid, message=message)
            try:
                self.repo.import_(
                    patches=io.BytesIO(patch.patch.encode("utf-8")),
                    message=message.encode("utf-8"),
                    user=user.encode("utf-8"),
                )
            except Exception as e:
                logger.info(
                    "Failed to apply patch: {}".format(e),
                    phid=patch.phid,
                    exc_info=True,
                )
                raise