Call command

[noissue]

pulpcore/cli/common/context.py

@@ -123,15 +123,22 @@ def output_result(self, result: Any) -> None:
                 _("Format '{format}' not implemented.").format(format=self.format)
             )
 
-    def call(self, operation_id: str, non_blocking: bool = False, *args: Any, **kwargs: Any) -> Any:
+    def call(
+        self,
+        operation_id: str,
+        non_blocking: bool = False,
+        parameters: Optional[Dict[str, Any]] = None,
+        body: Optional[Dict[str, Any]] = None,
+        uploads: Optional[Dict[str, bytes]] = None,
+    ) -> Any:
         """
         Perform an API call for operation_id.
         Wait for triggered tasks to finish if not background.
         Returns the operation result, or the finished task.
         If non_blocking, returns unfinished tasks.
         """
         try:
-            result = self.api.call(operation_id, *args, **kwargs)
+            result = self.api.call(operation_id, parameters=parameters, body=body, uploads=uploads)
         except OpenAPIError as e:
             raise click.ClickException(str(e))
         except HTTPError as e:

pulpcore/cli/common/debug.py

@@ -1,10 +1,11 @@
 import gettext
 import sys
-from typing import Optional
+from typing import IO, Any, Dict, Iterable, Optional
 
 import click
 
 from pulpcore.cli.common.context import PluginRequirement, PulpContext, pass_pulp_context
+from pulpcore.cli.common.generic import load_json_callback
 
 _ = gettext.gettext
 
@@ -66,7 +67,7 @@ def schema(pulp_ctx: PulpContext) -> None:
 
 
 @openapi_group.command()
-@click.option("--id", "operation_id", required=True)
+@click.option("--id", "operation_id", required=True, help=_("Operation ID in openapi schema"))
 @pass_pulp_context
 def operation(pulp_ctx: PulpContext, operation_id: str) -> None:
     """
@@ -91,4 +92,34 @@ def operation(pulp_ctx: PulpContext, operation_id: str) -> None:
 @openapi_group.command()
 @pass_pulp_context
 def operation_ids(pulp_ctx: PulpContext) -> None:
+    """
+    Print a list of available operation-ids.
+    """
     pulp_ctx.output_result(list(pulp_ctx.api.operations.keys()))
+
+
+@openapi_group.command()
+@click.option("--id", "operation_id", required=True, help=_("Operation ID in openapi schema"))
+@click.option("--parameter", "parameters", multiple=True)
+@click.option("--body", callback=load_json_callback)
+@click.option("--upload", "uploads", type=click.File("rb"), multiple=True)
+@pass_pulp_context
+def call(
+    pulp_ctx: PulpContext,
+    operation_id: str,
+    parameters: Iterable[str],
+    body: Any,
+    uploads: Iterable[IO[bytes]],
+) -> None:
+    """
+    Make a REST call by operation-id.
+
+    WARNING: Danger ahead!
+    """
+    try:
+        params: Dict[str, str] = dict(parameter.partition("=")[::2] for parameter in parameters)
+    except ValueError:
+        raise click.ClickException("Parameters must be in the form <key>=<value>.")
+    uploads_dict: Dict[str, bytes] = {file.name: file.read() for file in uploads}
+    result = pulp_ctx.call(operation_id, parameters=params, body=body, uploads=uploads_dict)
+    pulp_ctx.output_result(result)

pulpcore/cli/common/openapi.py

@@ -166,9 +166,12 @@ def render_request(
         uploads: Optional[Dict[str, bytes]] = None,
     ) -> requests.PreparedRequest:
         method_spec = path_spec[method]
-        content_types: List[str] = (
-            list(method_spec["requestBody"]["content"].keys()) if body or uploads else []
-        )
+        try:
+            content_types: List[str] = (
+                list(method_spec["requestBody"]["content"].keys()) if body or uploads else []
+            )
+        except KeyError:
+            raise OpenAPIError(_("This operation does not expect a request body."))
 
         data: Optional[Dict[str, Any]] = None
         json: Optional[Dict[str, Any]] = None