refactor(cli): move coverage reporting to custom_clusterlib

Replaced `clusterlib` with `custom_clusterlib` for CLI coverage
recording. Introduced a new `record_cli_coverage` function in
`custom_clusterlib` to handle CLI coverage tracking.

Author: mkoura

cardano_node_tests/cardano_cli_coverage.py

@@ -10,8 +10,7 @@
 import sys
 import typing as tp
 
-from cardano_clusterlib import clusterlib
-
+from cardano_node_tests.utils import custom_clusterlib
 from cardano_node_tests.utils import helpers
 
 LOGGER = logging.getLogger(__name__)
@@ -179,7 +178,9 @@ def get_log_coverage(log_file: pl.Path) -> dict:
         for line in infile:
             if not line.startswith("cardano-cli"):
                 continue
-            clusterlib.record_cli_coverage(cli_args=line.split(), coverage_dict=coverage_dict)
+            custom_clusterlib.record_cli_coverage(
+                cli_args=line.split(), coverage_dict=coverage_dict
+            )
 
     return coverage_dict
 

cardano_node_tests/utils/artifacts.py

@@ -2,7 +2,6 @@
 
 import json
 import logging
-import os
 import pathlib as pl
 import shutil
 
@@ -21,7 +20,7 @@
 def save_cli_coverage(cluster_obj: clusterlib.ClusterLib, pytest_config: Config) -> pl.Path | None:
     """Save CLI coverage info."""
     cli_coverage_dir = pytest_config.getoption(CLI_COVERAGE_ARG)
-    if not (cli_coverage_dir and cluster_obj.cli_coverage):
+    if not (cli_coverage_dir and hasattr(cluster_obj, "cli_coverage") and cluster_obj.cli_coverage):
         return None
 
     json_file = (
@@ -77,7 +76,7 @@ def save_cluster_artifacts(save_dir: pl.Path, state_dir: pl.Path) -> None:
             continue
         shutil.copytree(src_dir, destdir / dname, symlinks=True, ignore_dangling_symlinks=True)
 
-    if not os.listdir(destdir):
+    if not destdir.iterdir():
         destdir.rmdir()
         return
 

cardano_node_tests/utils/clusterlib_utils.py

@@ -15,6 +15,7 @@
 from cardano_clusterlib import clusterlib
 from cardano_clusterlib import txtools as cl_txtools
 
+from cardano_node_tests.utils import custom_clusterlib
 from cardano_node_tests.utils import helpers
 from cardano_node_tests.utils import submit_utils
 from cardano_node_tests.utils.faucet import fund_from_faucet  # noqa: F401 # for compatibility
@@ -945,9 +946,10 @@ def _get_ledger_state_cmd(
     ledger_state_cmd = " ".join(cardano_cli_args)
 
     # Record cli coverage
-    clusterlib.record_cli_coverage(
-        cli_args=cardano_cli_args, coverage_dict=cluster_obj.cli_coverage
-    )
+    if hasattr(cluster_obj, "cli_coverage"):
+        custom_clusterlib.record_cli_coverage(
+            cli_args=cardano_cli_args, coverage_dict=cluster_obj.cli_coverage
+        )
 
     return ledger_state_cmd
 

cardano_node_tests/utils/custom_clusterlib.py

@@ -3,13 +3,50 @@
 import logging
 import os
 import pathlib as pl
+import typing as tp
 
 from cardano_clusterlib import clusterlib
+from cardano_clusterlib import consts
 from cardano_clusterlib import transaction_group
 
 LOGGER = logging.getLogger(__name__)
 
 
+def record_cli_coverage(cli_args: list[str], coverage_dict: dict) -> None:
+    """Record coverage info for CLI commands.
+
+    Args:
+        cli_args: A list of command and it's arguments.
+        coverage_dict: A dictionary with coverage info.
+    """
+    parent_dict = coverage_dict
+    prev_arg = ""
+    for arg in cli_args:
+        # 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
+        if not cur_dict:
+            parent_dict[arg] = {"_count": 0}
+            cur_dict = parent_dict[arg]
+
+        # Increment count
+        cur_dict["_count"] += 1
+
+        # Set new parent dict
+        if not arg.startswith("--"):
+            parent_dict = cur_dict
+
+
 def create_submitted_file(tx_file: clusterlib.FileType) -> None:
     """Create a `.submitted` status file when the Tx was successfully submitted."""
     tx_path = pl.Path(tx_file)
@@ -23,13 +60,53 @@ def create_submitted_file(tx_file: clusterlib.FileType) -> None:
 
 
 class ClusterLib(clusterlib.ClusterLib):
+    def __init__(
+        self,
+        state_dir: clusterlib.FileType,
+        slots_offset: int | None = None,
+        socket_path: clusterlib.FileType = "",
+        command_era: str = consts.CommandEras.LATEST,
+    ):
+        super().__init__(
+            state_dir=state_dir,
+            slots_offset=slots_offset,
+            socket_path=socket_path,
+            command_era=command_era,
+        )
+        self.cli_coverage: dict[str, tp.Any] = {}
+
     @property
     def g_transaction(self) -> transaction_group.TransactionGroup:
         """Transaction group."""
         if not self._transaction_group:
             self._transaction_group = TransactionGroup(clusterlib_obj=self)
         return self._transaction_group
 
+    def cli(
+        self,
+        cli_args: list[str],
+        timeout: float | None = None,
+        add_default_args: bool = True,
+    ) -> clusterlib.CLIOut:
+        """Run the `cardano-cli` command.
+
+        Args:
+            cli_args: A list of arguments for cardano-cli.
+            timeout: A timeout for the command, in seconds (optional).
+            add_default_args: Whether to add default arguments to the command (optional).
+
+        Returns:
+            clusterlib.CLIOut: A data container containing command stdout and stderr.
+        """
+        cli_args_strs_all = [str(arg) for arg in cli_args]
+        if add_default_args:
+            cli_args_strs_all.insert(0, "cardano-cli")
+            cli_args_strs_all.insert(1, self.command_era)
+
+        record_cli_coverage(cli_args=cli_args_strs_all, coverage_dict=self.cli_coverage)
+
+        return super().cli(cli_args=cli_args, timeout=timeout, add_default_args=add_default_args)
+
 
 class TransactionGroup(transaction_group.TransactionGroup):
     def submit_tx(