Implement 'git-obs pr dump' command to store pull request information on disk

osc/commands_git/pr_dump.py

@@ -0,0 +1,182 @@
+import os
+from typing import Optional
+
+import osc.commandline_git
+
+
+class PullRequestDumpCommand(osc.commandline_git.GitObsCommand):
+    """
+    Dump a pull request to disk
+    """
+
+    name = "dump"
+    parent = "PullRequestCommand"
+
+    def init_arguments(self):
+        from osc.commandline_git import complete_checkout_pr
+
+        self.add_argument(
+            "id",
+            nargs="+",
+            help="Pull request ID in <owner>/<repo>#<number> format",
+        ).completer = complete_checkout_pr
+
+    def clone_or_update(
+        self,
+        owner: str,
+        repo: str,
+        *,
+        pr_number: Optional[int] = None,
+        branch: Optional[str] = None,
+        commit: str,
+        directory: str,
+        reference: Optional[str] = None,
+    ):
+        from osc import gitea_api
+
+        if not pr_number and not branch:
+            raise ValueError("Either 'pr_number' or 'branch' must be specified")
+
+        if not os.path.isdir(os.path.join(directory, ".git")):
+            gitea_api.Repo.clone(
+                self.gitea_conn,
+                owner,
+                repo,
+                directory=directory,
+                add_remotes=True,
+                reference=reference,
+            )
+
+        git = gitea_api.Git(directory)
+        git_owner, git_repo = git.get_owner_repo()
+        assert git_owner == owner, f"owner does not match: {git_owner} != {owner}"
+        assert git_repo == repo, f"repo does not match: {git_repo} != {repo}"
+
+        git.fetch()
+
+        if pr_number:
+            # checkout the pull request and check if HEAD matches head/sha from Gitea
+            pr_branch = git.fetch_pull_request(pr_number, force=True)
+            git.switch(pr_branch)
+            head_commit = git.get_branch_head()
+            assert head_commit == commit, f"HEAD of the current branch '{pr_branch}' is '{head_commit}' but the Gitea pull request points to '{commit}'"
+        elif branch:
+            git.switch(branch)
+            if not git.branch_contains_commit(commit=commit):
+                raise RuntimeError(f"Branch '{branch}' doesn't contain commit '{commit}'")
+            git.reset(commit, hard=True)
+        else:
+            raise ValueError("Either 'pr_number' or 'branch' must be specified")
+
+    def run(self, args):
+        import json
+        from osc import gitea_api
+        from osc import obs_api
+        from osc.util.xml import xml_indent
+        from osc.util.xml import ET
+
+        self.print_gitea_settings()
+
+        pull_request_ids = args.id
+
+        for pr_id in pull_request_ids:
+            owner, repo, number = gitea_api.PullRequest.split_id(pr_id)
+            pr_obj = gitea_api.PullRequest.get(self.gitea_conn, owner, repo, number)
+            path = os.path.join(owner, repo, str(number))
+
+            review_obj_list = pr_obj.get_reviews(self.gitea_conn)
+
+            # see https://github.com/go-gitea/gitea/blob/main/modules/structs/pull_review.go - look for "type ReviewStateType string"
+            state_map = {
+                "APPROVED": "accepted",
+                "REQUEST_CHANGES": "declined",
+                "REQUEST_REVIEW": "new",  # review hasn't started
+                "PENDING": "review",  # review is in progress
+                "COMMENT": "deleted",  # just to make XML validation happy, we'll replace it with "comment" later
+            }
+
+            xml_review_list = []
+            for review_obj in review_obj_list:
+                xml_review_list.append(
+                    {
+                        "state": state_map[review_obj.state],
+                        "who": review_obj.who,
+                        "created": review_obj.submitted_at,
+                        "when": review_obj.updated_at,
+                        "comment": review_obj.body,
+                    }
+                )
+
+            req = obs_api.Request(
+                id=pr_id,
+                title=pr_obj.title,
+                description=pr_obj.body,
+                creator=pr_obj.user,
+                # each pull request maps to only one action
+                action_list=[
+                    {
+                        "type": "submit",
+                        "source": {
+                            "project": pr_obj.head_owner,
+                            "package": pr_obj.head_repo,
+                            "rev": pr_obj.head_commit,
+                        },
+                        "target": {
+                            "project": pr_obj.base_owner,
+                            "package": pr_obj.base_repo,
+                        },
+                    },
+                ],
+                review_list=xml_review_list,
+            )
+
+            # HACK: changes to request XML that are not compatible with OBS
+            req_xml = req.to_xml()
+
+            req_xml_action = req_xml.find("action")
+            assert req_xml_action is not None
+            req_xml_action.attrib["type"] = "gitea-pull-request"
+            req_xml_action.insert(0, ET.Comment("The type='gitea-pull-request' attribute value is a custom extension to the OBS XML schema."))
+
+            req_xml_action_source = req_xml_action.find("source")
+            assert req_xml_action_source is not None
+            req_xml_action_source.append(ET.Comment("The 'branch' attribute is a custom extension to the OBS XML schema."))
+            req_xml_action_source.attrib["branch"] = pr_obj.head_branch
+
+            req_xml_action_target = req_xml_action.find("target")
+            assert req_xml_action_target is not None
+            req_xml_action_target.append(ET.Comment("The 'rev' and 'branch' attributes are custom extensions to the OBS XML schema."))
+            req_xml_action_target.attrib["rev"] = pr_obj.base_commit
+            req_xml_action_target.attrib["branch"] = pr_obj.base_branch
+
+            req_xml_review_list = req_xml.findall("review")
+            for req_xml_review in req_xml_review_list:
+                if req_xml_review.attrib["state"] == "deleted":
+                    req_xml_review.attrib["state"] = "comment"
+                    req_xml_review.insert(0, ET.Comment("The state='comment' attribute value is a custom extension to the OBS XML schema."))
+
+            metadata_dir = os.path.join(path, "metadata")
+            os.makedirs(metadata_dir, exist_ok=True)
+
+            with open(os.path.join(metadata_dir, "obs-request.xml"), "wb") as f:
+                xml_indent(req_xml)
+                ET.ElementTree(req_xml).write(f, encoding="utf-8")
+
+            with open(os.path.join(metadata_dir, "pr.json"), "w", encoding="utf-8") as f:
+                json.dump(pr_obj._data, f, indent=4, sort_keys=True)
+
+            with open(os.path.join(metadata_dir, "base.json"), "w", encoding="utf-8") as f:
+                json.dump(pr_obj._data["base"], f, indent=4, sort_keys=True)
+
+            with open(os.path.join(metadata_dir, "head.json"), "w", encoding="utf-8") as f:
+                json.dump(pr_obj._data["head"], f, indent=4, sort_keys=True)
+
+            with open(os.path.join(metadata_dir, "reviews.json"), "w", encoding="utf-8") as f:
+                json.dump([i._data for i in review_obj_list], f, indent=4, sort_keys=True)
+
+            base_dir = os.path.join(path, "base")
+            # we must use the `merge_base` instead of `head_commit`, because the latter changes after merging the PR and the `base` directory would contain incorrect data
+            self.clone_or_update(owner, repo, branch=pr_obj.base_branch, commit=pr_obj.merge_base, directory=base_dir)
+
+            head_dir = os.path.join(path, "head")
+            self.clone_or_update(owner, repo, pr_number=pr_obj.number, commit=pr_obj.head_commit, directory=head_dir, reference=base_dir)