def unlanded_wpt_commits_by_pr()

in sync/landing.py [0:0]

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

def unlanded_wpt_commits_by_pr(git_gecko: Repo,
                               git_wpt: Repo,
                               prev_wpt_head: str,
                               wpt_head: str = "origin/master",
                               ) -> list[tuple[int | None, list[WptCommit]]]:
    revish = f"{prev_wpt_head}..{wpt_head}"

    commits_by_pr: list[tuple[int | None, list[WptCommit]]] = []
    index_by_pr: dict[int | None, int] = {}

    for commit in git_wpt.iter_commits(revish,
                                       reverse=True,
                                       first_parent=True):
        wpt_commit = sync_commit.WptCommit(git_wpt, commit.hexsha)
        pr = wpt_commit.pr()
        extra_commits = []
        if pr not in index_by_pr:
            pr_data: tuple[int | None, list[WptCommit]] = (pr, [])
            # If we have a merge commit, also get the commits merged in
            if len(commit.parents) > 1:
                merged_revish = f"{commit.parents[0].hexsha}..{commit.hexsha}"
                for merged_commit in git_wpt.iter_commits(merged_revish,
                                                          reverse=True):
                    if merged_commit.hexsha != commit.hexsha:
                        wpt_commit = sync_commit.WptCommit(git_wpt, merged_commit.hexsha)
                        if wpt_commit.pr() == pr:
                            extra_commits.append(wpt_commit)
        else:
            idx = index_by_pr[pr]
            pr_data = commits_by_pr.pop(idx)
            assert pr_data[0] == pr
            index_by_pr = {key: (value if value < idx else value - 1)
                           for key, value in index_by_pr.items()}
        for c in extra_commits + [wpt_commit]:
            pr_data[1].append(c)
        commits_by_pr.append(pr_data)
        index_by_pr[pr] = len(commits_by_pr) - 1

    return commits_by_pr