Merge pull request #167 from input-output-hk/fix_subcommand_args

Distinguish between arguments for nested subcommands

cardano_clusterlib/clusterlib_klass.py

@@ -173,14 +173,15 @@ def cli(self, cli_args: List[str]) -> structs.CLIOut:
         Returns:
             structs.CLIOut: A tuple containing command stdout and stderr.
         """
-        cli_args_strs = [str(arg) for arg in cli_args]
-        cli_args_strs.insert(0, "cardano-cli")
+        cli_args_strs_all = [str(arg) for arg in cli_args]
+        cli_args_strs_all.insert(0, "cardano-cli")
+        cli_args_strs = [arg for arg in cli_args_strs_all if arg != consts.SUBCOMMAND_MARK]
 
         cmd_str = clusterlib_helpers._format_cli_args(cli_args=cli_args_strs)
         clusterlib_helpers._write_cli_log(clusterlib_obj=self, command=cmd_str)
         LOGGER.debug("Running `%s`", cmd_str)
 
-        coverage.record_cli_coverage(cli_args=cli_args_strs, coverage_dict=self.cli_coverage)
+        coverage.record_cli_coverage(cli_args=cli_args_strs_all, coverage_dict=self.cli_coverage)
 
         # re-run the command when running into
         # Network.Socket.connect: <socket: X>: resource exhausted (Resource temporarily unavailable)

cardano_clusterlib/consts.py

@@ -9,6 +9,14 @@
 }
 
 
+# The SUBCOMMAND_MARK is used to mark the beginning of a subcommand. It is used to differentiate
+# between options and subcommands. That is needed for CLI coverage recording.
+# For example, the command `cardano-cli query tx-mempool --cardano-mode info`
+# has the following arguments:
+#  ["query", "tx-mempool", "--cardano-mode", SUBCOMMAND_MARK, "info"]
+SUBCOMMAND_MARK = "SUBCOMMAND"
+
+
 class Protocols:
     CARDANO = "cardano"
     SHELLEY = "shelley"

cardano_clusterlib/coverage.py

@@ -1,5 +1,7 @@
 from typing import List
 
+from cardano_clusterlib import consts
+
 
 def record_cli_coverage(cli_args: List[str], coverage_dict: dict) -> None:
     """Record coverage info for CLI commands.
@@ -11,20 +13,26 @@ def record_cli_coverage(cli_args: List[str], coverage_dict: dict) -> None:
     parent_dict = coverage_dict
     prev_arg = ""
     for arg in cli_args:
-        # if the current argument is a parameter to an option, skip it
+        # If the current argument is a subcommand marker, record it and skip it
+        if arg == consts.SUBCOMMAND_MARK:
+            prev_arg = arg
+            continue
+
+        # If the current argument is a parameter to an option, skip it
         if prev_arg.startswith("--") and not arg.startswith("--"):
             continue
+
         prev_arg = arg
 
         cur_dict = parent_dict.get(arg)
-        # initialize record if it doesn't exist yet
+        # Initialize record if it doesn't exist yet
         if not cur_dict:
             parent_dict[arg] = {"_count": 0}
             cur_dict = parent_dict[arg]
 
-        # increment count
+        # Increment count
         cur_dict["_count"] += 1
 
-        # set new parent dict
+        # Set new parent dict
         if not arg.startswith("--"):
             parent_dict = cur_dict

cardano_clusterlib/query_group.py

@@ -31,14 +31,15 @@ class QueryGroup:
     def __init__(self, clusterlib_obj: "types.ClusterLib") -> None:
         self._clusterlib_obj = clusterlib_obj
 
-    def query_cli(self, cli_args: UnpackableSequence) -> str:
+    def query_cli(self, cli_args: UnpackableSequence, cli_sub_args: UnpackableSequence = ()) -> str:
         """Run the `cardano-cli query` command."""
         stdout = self._clusterlib_obj.cli(
             [
                 "query",
                 *cli_args,
                 *self._clusterlib_obj.magic_args,
                 f"--{self._clusterlib_obj.protocol}-mode",
+                *cli_sub_args,
             ]
         ).stdout
         stdout_dec = stdout.decode("utf-8") if stdout else ""
@@ -453,7 +454,9 @@ def get_mempool_info(self) -> Dict[str, Any]:
         Returns:
             dict: A dictionary containing mempool information.
         """
-        tx_mempool: Dict[str, Any] = json.loads(self.query_cli(["tx-mempool", "info"]))
+        tx_mempool: Dict[str, Any] = json.loads(
+            self.query_cli(["tx-mempool"], cli_sub_args=[consts.SUBCOMMAND_MARK, "info"])
+        )
         return tx_mempool
 
     def get_mempool_next_tx(self) -> Dict[str, Any]:
@@ -462,7 +465,9 @@ def get_mempool_next_tx(self) -> Dict[str, Any]:
         Returns:
             dict: A dictionary containing mempool information.
         """
-        tx_mempool: Dict[str, Any] = json.loads(self.query_cli(["tx-mempool", "next-tx"]))
+        tx_mempool: Dict[str, Any] = json.loads(
+            self.query_cli(["tx-mempool"], cli_sub_args=[consts.SUBCOMMAND_MARK, "next-tx"])
+        )
         return tx_mempool
 
     def get_mempool_tx_exists(self, txid: str) -> Dict[str, Any]:
@@ -474,7 +479,9 @@ def get_mempool_tx_exists(self, txid: str) -> Dict[str, Any]:
         Returns:
             dict: A dictionary containing mempool information.
         """
-        tx_mempool: Dict[str, Any] = json.loads(self.query_cli(["tx-mempool", "tx-exists", txid]))
+        tx_mempool: Dict[str, Any] = json.loads(
+            self.query_cli(["tx-mempool"], cli_sub_args=[consts.SUBCOMMAND_MARK, "tx-exists", txid])
+        )
         return tx_mempool
 
     def __repr__(self) -> str: