Add '--timeline' option to 'git-obs pr get'

osc/gitea_api/issue_timeline_entry.py

@@ -0,0 +1,250 @@
+from typing import List
+from typing import Optional
+
+from .common import GiteaModel
+from .common import dt_sanitize
+from .connection import Connection
+from .pr_review import PullRequestReview
+from .user import User
+
+
+class IssueTimelineEntry(GiteaModel):
+    @property
+    def type(self) -> str:
+        return self._data["type"]
+
+    @property
+    def body(self) -> str:
+        return self._data["body"]
+
+    @property
+    def user(self) -> str:
+        return self._data["user"]["login"]
+
+    @property
+    def user_obj(self) -> User:
+        return User(self._data["user"], response=self._response)
+
+    @property
+    def created_at(self) -> str:
+        return self._data["created_at"]
+
+    @property
+    def updated_at(self) -> str:
+        return self._data["updated_at"]
+
+    @property
+    def created_updated_str(self) -> str:
+        result = dt_sanitize(self.created_at)
+        if self.updated_at and self.updated_at != self.created_at:
+            result += f" (updated: {dt_sanitize(self.updated_at)})"
+        return result
+
+    @property
+    def review_id(self) -> Optional[int]:
+        return self._data["review_id"]
+
+    @property
+    def review_obj(self) -> Optional[PullRequestReview]:
+        from .exceptions import PullRequestReviewDoesNotExist
+
+        if not self.review_id:
+            return None
+        try:
+            return PullRequestReview.get(self._conn, self.pr_owner, self.pr_repo, self.pr_number, str(self.review_id))
+        except PullRequestReviewDoesNotExist:
+            # reviews can be removed from the database, but their IDs remain in other places
+            return None
+
+    @property
+    def pr_owner(self) -> str:
+        from .pr import PullRequest
+
+        return PullRequest.get_owner_repo_number(self._data["pull_request_url"])[0]
+
+    @property
+    def pr_repo(self) -> str:
+        from .pr import PullRequest
+
+        return PullRequest.get_owner_repo_number(self._data["pull_request_url"])[1]
+
+    @property
+    def pr_number(self) -> int:
+        from .pr import PullRequest
+
+        return PullRequest.get_owner_repo_number(self._data["pull_request_url"])[2]
+
+    def format(self):
+        handler = getattr(self, f"_format_{self.type}", None)
+        if handler is None:
+            return (self.type, self.body)
+        if not callable(handler):
+            raise TypeError(f"Handler for {self.type} is not callable")
+        return handler()  # pylint: disable=not-callable
+
+    def _format_assignees(self):
+        if self._data["removed_assignee"]:
+            return f"unassigned the pull request from {self._data['assignee']['login']}", None
+        return f"assigned the pull request to {self._data['assignee']['login']}", None
+
+    def _format_change_target_branch(self):
+        return f"changed target branch from '{self._data['old_ref']}' to '{self._data['new_ref']}'", None
+
+    def _format_change_title(self):
+        return "changed title", f"from '{self._data['old_title']}' to '{self._data['new_title']}'"
+
+    def _format_comment(self):
+        return "commented", self.body
+
+    def _format_comment_ref(self):
+        return "referenced the pull request", self._data["ref_comment"]["html_url"]
+
+    def _format_commit_ref(self):
+        import urllib.parse
+        from osc.util import xml as osc_xml
+
+        node = osc_xml.xml_fromstring(self.body)
+        assert node.tag == "a"
+
+        netloc = self._conn.host
+        if self._conn.port:
+            netloc += f":{self._conn.port}"
+        url = urllib.parse.urlunsplit((self._conn.scheme, netloc, node.attrib["href"], "", ""))
+        body = f"{url}\n{node.text}".strip()
+
+        return f"referenced the pull request from commit", body
+
+    def _format_close(self):
+        return "closed the pull request", self.body
+
+    def _format_delete_branch(self):
+        return f"deleted branch '{self._data['old_ref']}'", None
+
+    def _format_dismiss_review(self):
+        return f"dismissed {self.review_obj.user}'s review", self.body
+
+    def _format_merge_pull(self):
+        from .pr import PullRequest
+
+        pr_obj = PullRequest.get(self._conn, self.pr_owner, self.pr_repo, self.pr_number)
+        return f"merged commit {pr_obj.merge_commit} to {pr_obj.base_branch}", None
+
+    def _format_pull_cancel_scheduled_merge(self):
+        return "canceled auto merging the pull request when all checks succeed", None
+
+    def _format_pull_push(self):
+        import json
+
+        data = json.loads(self.body)
+        len_commits = len(data["commit_ids"])
+        return f"{'force-' if data['is_force_push'] else ''}pushed {len_commits} commit{'s' if len_commits > 1 else ''}", None
+
+    def _format_pull_ref(self):
+        return "referenced the pull request", f"{self._data['ref_issue']['html_url']}\n{self._data['ref_issue']['title']}"
+
+    def _format_pull_scheduled_merge(self):
+        return "scheduled the pull request to auto merge when all checks succeed", None
+
+    def _format_reopen(self):
+        return "reopened the pull request", self.body
+
+    def _format_review(self):
+        messages = {
+            "APPROVED": "approved",
+            "REQUEST_CHANGES": "declined",
+            "COMMENTED": "commented",
+        }
+        msg = messages.get(self.review_obj.state, self.review_obj.state)
+        return f"{msg} the review", self.body
+
+    def _format_review_request(self):
+        reviewer = self._data["assignee"]["login"] if  self._data["assignee"] else self._data["assignee_team"]["name"]
+        return f"requested review from {reviewer}", self.body
+
+    # unused; we are not interested in these types of entries
+
+    def _format_added_deadline(self):
+        return None, None
+
+    def _format_modified_deadline(self):
+        return None, None
+
+    def _format_removed_deadline(self):
+        return None, None
+
+    def _format_pin(self):
+        return None, None
+
+    def _format_unpin(self):
+        return None, None
+
+    def _format_change_time_estimate(self):
+        return None, None
+
+    def _format_project(self):
+        return None, None
+
+    def _format_project_board(self):
+        return None, None
+
+    def _format_start_tracking(self):
+        return None, None
+
+    def _format_stop_tracking(self):
+        return None, None
+
+    def _format_add_time_manual(self):
+        return None, None
+
+    def _format_delete_time_manual(self):
+        return None, None
+
+    def _format_cancel_tracking(self):
+        return None, None
+
+    def _format_label(self):
+        return None, None
+
+    def _format_milestone(self):
+        return None, None
+
+    def _format_lock(self):
+        return None, None
+
+    def _format_unlock(self):
+        return None, None
+
+    def _format_add_dependency(self):
+        return None, None
+
+    def _format_remove_dependency(self):
+        return None, None
+
+    # TODO: find a reproducer for formatting the following entries
+    # def _format_issue_ref(self):
+    # def _format_code(self):
+    # def _format_change_issue_ref(self):
+
+    @classmethod
+    def list(
+        cls,
+        conn: Connection,
+        owner: str,
+        repo: str,
+        number: int,
+    ) -> List["IssueTimelineEntry"]:
+        """
+        List issue timeline entries (applicable to issues and pull request).
+
+        :param conn: Gitea ``Connection`` instance.
+        :param owner: Owner of the repo.
+        :param repo: Name of the repo.
+        :param number: Number of the pull request in owner/repo.
+        """
+        q = {
+            "limit": -1,
+        }
+        url = conn.makeurl("repos", owner, repo, "issues", str(number), "timeline", query=q)
+        response = conn.request("GET", url)
+        obj_list = [cls(i, response=response, conn=conn) for i in response.json() or []]
+        return obj_list