def update_commits()

in sync/downstream.py [0:0]

• 83 lines of code
• 7 McCabe index (conditional complexity)

    def update_commits(self) -> bool:
        exception = None
        try:
            self.update_wpt_commits()

            # Check if this sync reverts some unlanded earlier PR and if so mark both
            # as skip and don't try to apply the commits here
            reverts = self.reverts_syncs()
            if reverts:
                all_open = all(item.status == "open" for item in reverts)
                for revert_sync in reverts:
                    if revert_sync.status == "open":
                        logger.info("Skipping sync for PR %s because it is later reverted" %
                                    revert_sync.pr)
                        with SyncLock.for_process(revert_sync.process_name) as revert_lock:
                            assert isinstance(revert_lock, SyncLock)
                            with revert_sync.as_mut(revert_lock):
                                revert_sync.skip = True  # type: ignore
                # TODO: If this commit reverts some closed syncs, then set the metadata
                # commit of this commit to the revert of the metadata commit from that
                # sync
                if all_open:
                    logger.info("Sync was a revert of other open syncs, skipping")
                    self.skip = True  # type: ignore
                    return False

            old_gecko_head = self.gecko_commits.head.sha1
            logger.debug(f"PR {self.pr} gecko HEAD was {old_gecko_head}")

            def plain_apply() -> bool:
                logger.info("Applying on top of the current commits")
                self.wpt_to_gecko_commits()
                return True

            def rebase_apply() -> bool:
                logger.info("Applying with a rebase onto latest integration branch")
                new_base = self.gecko_integration_branch()
                gecko_work = self.gecko_worktree.get()
                reset_head = "HEAD"
                if (len(self.gecko_commits) > 0 and
                    self.gecko_commits[0].metadata.get("wpt-type") == "dependent"):
                    # If we have any dependent commits first reset to the new
                    # head. This prevents conflicts if the dependents already
                    # landed
                    # TODO: Actually check if they landed?
                    reset_head = new_base
                gecko_work.git.reset(reset_head, hard=True)
                self.gecko_rebase(new_base, abort_on_fail=True)
                self.wpt_to_gecko_commits()
                return True

            def dependents_apply() -> bool:
                logger.info("Applying with upstream dependents")
                dependencies = self.unlanded_commits_same_files()
                if dependencies:
                    logger.info("Found dependencies:\n%s" %
                                "\n".join(item.msg.splitlines()[0].decode("utf8", "replace")
                                          for item in dependencies))
                    self.wpt_to_gecko_commits(dependencies)
                    assert self.bug is not None
                    env.bz.comment(self.bug,
                                   "PR %s applied with additional changes from upstream: %s"
                                   % (self.pr, ", ".join(item.sha1 for item in dependencies)))
                    return True
                return False

            error = None
            for fn in [plain_apply, rebase_apply, dependents_apply]:
                try:
                    if fn():
                        error = None
                        break
                    else:
                        logger.error("Applying with %s was a no-op" % fn.__name__)
                except Exception as e:
                    import traceback
                    error = e
                    logger.error("Applying with %s errored" % fn.__name__)
                    logger.error(traceback.format_exc())

            if error is not None:
                raise error

            logger.debug(f"PR {self.pr} gecko HEAD now {self.gecko_commits.head.sha1}")
            if old_gecko_head == self.gecko_commits.head.sha1:
                logger.info("Gecko commits did not change for PR %s" % self.pr)
                return False

            # If we have a metadata commit already, ensure it's applied now
            if "metadata-commit" in self.data:
                self.ensure_metadata_commit()

            renames = self.wpt_renames()
            self.move_metadata(renames)
            self.update_bug_components(renames)

            files_changed = self.files_changed()
            self.set_bug_component(files_changed)
        except Exception as e:
            exception = e
            raise
        finally:
            # If we managed to apply all the commits without error, reset the error flag
            # otherwise update it with the current exception
            self.error = exception
        return True