def reverts_syncs()

in sync/downstream.py [0:0]

• 34 lines of code
• 18 McCabe index (conditional complexity)

    def reverts_syncs(self) -> set[DownstreamSync]:
        """Return a set containing the previous syncs reverted by this one, if any"""
        revert_re = re.compile(b"This reverts commit ([0-9A-Fa-f]+)")
        unreverted_commits = defaultdict(set)
        for commit in self.wpt_commits:
            if not commit.msg.startswith(b"Revert "):
                # If not everything is a revert then return
                return set()
            revert_shas = revert_re.findall(commit.msg)
            if len(revert_shas) == 0:
                return set()
            # Just use the first match for now
            sha = revert_shas[0].decode("ascii")
            try:
                # Reassign the hash here, in case a short hash was used for reverting.
                sha = str(self.git_wpt.rev_parse(sha))
            except (ValueError, git.BadName):
                # Commit isn't in this repo (could be upstream)
                return set()
            pr = env.gh_wpt.pr_for_commit(sha)
            if pr is None:
                return set()
            sync = DownstreamSync.for_pr(self.git_gecko, self.git_wpt, pr)
            if sync is None:
                return set()
            assert isinstance(sync, DownstreamSync)
            if sync not in unreverted_commits:
                # Ensure we have the latest commits for the reverted sync
                with SyncLock.for_process(sync.process_name) as revert_lock:
                    assert isinstance(revert_lock, SyncLock)
                    with sync.as_mut(revert_lock):
                        sync.update_wpt_commits()
                unreverted_commits[sync] = {item.sha1 for item in sync.wpt_commits}
            if sha in unreverted_commits[sync]:
                unreverted_commits[sync].remove(sha)
            # If the commit is not part of the sync, check if the PR was squashed and then reverted,
            # in that case all commits of the sync should be reverted.
            elif sha == env.gh_wpt.merge_sha(pr):
                unreverted_commits[sync] = set()

        rv = {sync for sync, unreverted in unreverted_commits.items()
              if not unreverted}
        return rv