Add 'git-obs staging search' command

osc/commands_git/staging_search.py

@@ -0,0 +1,108 @@
+import sys
+
+import osc.commandline_git
+
+
+class StagingSearchCommand(osc.commandline_git.GitObsCommand):
+    """
+    Search staging pull requests.
+    """
+
+    name = "search"
+    parent = "StagingCommand"
+
+    def init_arguments(self):
+        self.add_argument_owner_repo()
+        self.add_argument(
+            "--type",
+            dest="type",
+            # the choices must match the *_LABEL constants in StagingPullRequestWrapper class
+            choices=("BACKLOG", "INPROGRESS", "ONHOLD"),
+            required=True,
+            help="Filter by review state.",
+        )
+        self.add_argument(
+            "--package-review-state",
+            dest="package_review_state",
+            choices=("APPROVED", "ALL"),
+            default="APPROVED",
+            help="Filter by review state on *all* referenced *package* PRs.",
+        )
+        self.add_argument(
+            "--target-branch",
+            dest="target_branches",
+            action="append",
+            help="Filter by target branch.",
+        )
+        self.add_argument(
+            "--export",
+            action="store_true",
+            help="Show json objects instead of human readable text",
+        )
+
+    def run(self, args):
+        from osc import gitea_api
+        from osc.output import KeyValueTable
+        from osc.output import tty
+
+        self.print_gitea_settings()
+
+        pr_state = "open"
+        owner, repo = args.owner_repo
+
+        labels = gitea_api.Repo.get_label_ids(self.gitea_conn, owner, repo)
+
+        label = getattr(gitea_api.StagingPullRequestWrapper, f"{args.type}_LABEL")
+        label_id = labels.get(label, None)
+        if label_id is None:
+            raise gitea_api.GitObsRuntimeError(f"Label '{label}' doesn't exist in '{owner}/{repo}'")
+
+        pr_obj_list = gitea_api.PullRequest.list(self.gitea_conn, owner, repo, state=pr_state, labels=[label_id])
+        pr_obj_list.sort()
+
+        if args.target_branches:
+            pr_obj_list = [i for i in pr_obj_list if i.base_branch in args.target_branches]
+
+        table = KeyValueTable()
+        result = []
+        skipped = []
+        for pr in pr_obj_list:
+            ref_prs = pr.parse_pr_references()
+            if len(ref_prs) == 0:
+                skipped.append(pr)
+                continue
+
+            package_review_state_matched = True
+
+            if args.package_review_state != "ALL":
+                for ref_owner, ref_repo, ref_pr_number in ref_prs:
+                    ref_pr = gitea_api.PullRequest.get(self.gitea_conn, ref_owner, ref_repo, ref_pr_number)
+
+                    all_reviews = ref_pr.get_reviews(self.gitea_conn)
+                    for review_obj in all_reviews:
+                        if review_obj.state != args.package_review_state:
+                            package_review_state_matched = False
+                            break
+
+                    if not package_review_state_matched:
+                        break
+
+            if package_review_state_matched:
+                if args.export:
+                    result.append(pr.dict())
+                else:
+                    table.add(pr.id, pr.title)
+                    table.add("", pr.url)
+
+        if args.export:
+            from json import dumps
+
+            print(dumps(result, indent=4, sort_keys=True))
+        else:
+            print(str(table))
+
+        # print warnings at the end to make them more obvious
+        if skipped:
+            print(file=sys.stderr)
+            for pr_obj in skipped:
+                print(f"{tty.colorize('WARNING', 'yellow,bold')}: Skipped '{pr_obj.id}' due to empty or invalid 'PR:' references.", file=sys.stderr)

osc/gitea_api/pr.py

@@ -10,6 +10,7 @@
 from .connection import Connection
 from .connection import GiteaHTTPResponse
 from .user import User
+from .repo import Repo
 
 if typing.TYPE_CHECKING:
     from .pr_review import PullRequestReview
@@ -18,10 +19,10 @@
 @functools.total_ordering
 class PullRequest(GiteaModel):
     def __eq__(self, other):
-        (self.base_owner, self.base_repo, self.number) == (other.base_owner, other.base_repo, other.number)
+        return (self.base_owner, self.base_repo, self.number) == (other.base_owner, other.base_repo, other.number)
 
     def __lt__(self, other):
-        (self.base_owner, self.base_repo, self.number) < (other.base_owner, other.base_repo, other.number)
+        return (self.base_owner, self.base_repo, self.number) < (other.base_owner, other.base_repo, other.number)
 
     @classmethod
     def split_id(cls, pr_id: str) -> Tuple[str, str, int]:
@@ -363,7 +364,7 @@ def create(
         """
         url = conn.makeurl("repos", target_owner, target_repo, "pulls")
         if labels:
-            ids = cls._get_label_ids(conn, target_owner, target_repo)
+            ids = Repo.get_label_ids(conn, target_owner, target_repo)
             labels = [ids[i] for i in labels]
         data = {
             "base": target_branch,
@@ -438,6 +439,7 @@ def list(
         repo: str,
         *,
         state: Optional[str] = "open",
+        labels: Optional[List[int]] = None,
     ) -> List["PullRequest"]:
         """
         List pull requests in a repo.
@@ -454,6 +456,10 @@ def list(
             "state": state,
             "limit": 50,
         }
+
+        if labels:
+            q["labels"] = labels
+
         url = conn.makeurl("repos", owner, repo, "pulls", query=q)
         obj_list = []
         for response in conn.request_all_pages("GET", url):
@@ -717,19 +723,6 @@ def reopen(
         obj = cls(response.json(), response=response, conn=conn)
         return obj
 
-    @classmethod
-    def _get_label_ids(cls, conn: Connection, owner: str, repo: str) -> Dict[str, int]:
-        """
-        Helper to map labels to their IDs
-        """
-        result = {}
-        url = conn.makeurl("repos", owner, repo, "labels")
-        response = conn.request("GET", url)
-        labels = response.json()
-        for label in labels:
-            result[label["name"]] = label["id"]
-        return result
-
     @classmethod
     def add_labels(
         cls,
@@ -752,7 +745,7 @@ def add_labels(
 
         label_id_list = []
         invalid_labels = []
-        label_name_id_map = cls._get_label_ids(conn, owner, repo)
+        label_name_id_map = Repo.get_label_ids(conn, owner, repo)
         for label in labels:
             label_id = label_name_id_map.get(label, None)
             if not label_id:
@@ -791,7 +784,7 @@ def remove_labels(
 
         label_id_list = []
         invalid_labels = []
-        label_name_id_map = cls._get_label_ids(conn, owner, repo)
+        label_name_id_map = Repo.get_label_ids(conn, owner, repo)
         for label in labels:
             label_id = label_name_id_map.get(label, None)
             if not label_id:

osc/gitea_api/repo.py

@@ -2,7 +2,7 @@
 import os
 import re
 import subprocess
-from typing import List
+from typing import Dict, List
 from typing import Optional
 from typing import Tuple
 
@@ -312,3 +312,16 @@ def list_user_repos(cls, conn: Connection, owner: str) -> List["Repo"]:
         for response in conn.request_all_pages("GET", url):
             obj_list.extend([cls(i, response=response) for i in response.json()])
         return obj_list
+
+    @classmethod
+    def get_label_ids(cls, conn: Connection, owner: str, repo: str) -> Dict[str, int]:
+        """
+        Helper to map labels to their IDs
+        """
+        result = {}
+        url = conn.makeurl("repos", owner, repo, "labels")
+        response = conn.request("GET", url)
+        labels = response.json()
+        for label in labels:
+            result[label["name"]] = label["id"]
+        return result

osc/gitea_api/staging.py

@@ -5,7 +5,8 @@
 class StagingPullRequestWrapper:
     BACKLOG_LABEL = "staging/Backlog"
     INPROGRESS_LABEL = "staging/In Progress"
-
+    ONHOLD_LABEL = "staging/On Hold"
+
     def __init__(self, conn, owner: str, repo: str, number: int, *, topdir: str, cache_directory: Optional[str] = None):
         from . import PullRequest