update to gbpcli 2.x

pyproject.toml

@@ -6,7 +6,7 @@ authors = [
     {name = "Albert Hopkins", email = "[email protected]"},
 ]
 dependencies = [
-    "gbpcli>=1.7.0",
+    "gbpcli>=2.0.0",
     "rich>=13.6.0",
 ]
 requires-python = ">=3.11"

src/gbp_ps/cli/__init__.py

@@ -1,20 +1 @@
 """gbp-ps command-line interface"""
-from typing import cast
-
-from gbpcli import GBP
-from gbpcli.graphql import Query
-
-
-def get_dist_query(name: str, gbp: GBP, distribution: str = "gbp_ps") -> Query:
-    """Return the query with the given name
-
-    This function has a side-effect on pre 2.0 versions of gbpcli in that it mutates the
-    gbp argument to point to the given distribution's query database.
-    """
-    if hasattr(gbp.query, "_distribution"):
-        # Older GBP can only see the queries for the "gbpcli" distribution and need to
-        # be overridden to see queries from other distributions
-        gbp.query._distribution = distribution  # pylint: disable=protected-access
-        return cast(Query, getattr(gbp.query, name))
-
-    return cast(Query, getattr(getattr(gbp.query, distribution), name))

src/gbp_ps/cli/add_process.py

@@ -8,15 +8,13 @@
 
 from gbp_ps.types import BuildProcess
 
-from . import get_dist_query
-
 now = dt.datetime.now
 
 
 def handler(args: argparse.Namespace, gbp: GBP, _console: Console) -> int:
     """Show add/update an entry in the process table"""
     check(
-        get_dist_query("add_process", gbp)(
+        gbp.query.gbp_ps.add_process(
             process=BuildProcess(
                 build_host=platform.node(),
                 build_id=args.number,

src/gbp_ps/cli/ps.py

@@ -10,8 +10,6 @@
 from rich.live import Live
 from rich.table import Table
 
-from . import get_dist_query
-
 ModeHandler = Callable[[argparse.Namespace, Query, Console], int]
 ProcessList: TypeAlias = list[dict[str, Any]]
 
@@ -110,7 +108,7 @@ def handler(args: argparse.Namespace, gbp: GBP, console: Console) -> int:
     """Show currently building packages"""
     mode: ModeHandler = MODES[args.continuous]
 
-    return mode(args, get_dist_query("get_processes", gbp), console)
+    return mode(args, gbp.query.gbp_ps.get_processes, console)
 
 
 def parse_args(parser: argparse.ArgumentParser) -> None:

tests/test_cli.py

@@ -1,7 +1,6 @@
 """CLI unit tests for gbp-ps"""
 # pylint: disable=missing-docstring
 import datetime as dt
-import importlib.resources
 import io
 import platform
 from argparse import ArgumentParser, Namespace
@@ -11,14 +10,13 @@
 import rich.console
 from django.test.client import Client
 from gbpcli import GBP, Console
-from gbpcli.graphql import Query
 from gbpcli.theme import DEFAULT_THEME
 from requests import Response
 from requests.adapters import BaseAdapter
 from requests.structures import CaseInsensitiveDict
 from rich.theme import Theme
 
-from gbp_ps.cli import add_process, get_dist_query, ps
+from gbp_ps.cli import add_process, ps
 
 from . import LOCAL_TIMEZONE, TestCase, make_build_process
 
@@ -228,7 +226,7 @@ def test_continuous_mode(self, mock_sleep: mock.Mock) -> None:
 
         gbp = mock.Mock()
         mock_graphql_resp = [process.to_dict() for process in processes]
-        gbp.query.get_processes.side_effect = (
+        gbp.query.gbp_ps.get_processes.side_effect = (
             ({"buildProcesses": mock_graphql_resp}, None),
             KeyboardInterrupt,
         )
@@ -288,30 +286,3 @@ def test_parse_args(self) -> None:
         # Just ensure that parse_args is there and works
         parser = ArgumentParser()
         add_process.parse_args(parser)
-
-
-GET_PROCESSES_QUERY_STR = (
-    importlib.resources.files("gbp_ps") / "queries/get_processes.graphql"
-).read_text(encoding="UTF-8")
-
-
-class GetDistQueryTests(TestCase):
-    """Tests for the get_dist_query helper"""
-
-    def test_old_world(self) -> None:
-        # Pre gbpcli-2.0
-        gbp = test_gbp("http://test.invalid/")
-        query = get_dist_query("get_processes", gbp, distribution="gbp_ps")
-
-        self.assertEqual(str(query), GET_PROCESSES_QUERY_STR)
-
-    def test_new_world(self) -> None:
-        # post gbpcli-2.0. Not yet released so we mock the expected behavior
-        gbp = mock.MagicMock()
-        del gbp.query._distribution
-        gbp.query.gbp_ps.get_processes = Query(
-            GET_PROCESSES_QUERY_STR, "http://test.invalid", mock.Mock()
-        )
-        query = get_dist_query("get_processes", gbp, distribution="gbp_ps")
-
-        self.assertEqual(str(query), GET_PROCESSES_QUERY_STR)