feat: add script to get ADA amounts locked on testnet

- Introduced `testnet_cleanup_info.py` to fetch ADA amounts still locked.
- Added `info` and `info_addresses` functions in `testnet_cleanup.py` to
  calculate total balance and rewards of all addresses in a given location.

Author: mkoura

cardano_node_tests/testnet_cleanup_info.py

@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Print the Lovelace balance and rewards of all addresses in the given location."""
+
+import argparse
+import logging
+import os
+import pathlib as pl
+import sys
+
+from cardano_clusterlib import clusterlib
+
+from cardano_node_tests.utils import helpers
+from cardano_node_tests.utils import testnet_cleanup
+
+LOGGER = logging.getLogger(__name__)
+
+
+def get_args() -> argparse.Namespace:
+    """Get command line arguments."""
+    parser = argparse.ArgumentParser(description=__doc__.split("\n", maxsplit=1)[0])
+    parser.add_argument(
+        "-a",
+        "--artifacts-base-dir",
+        required=True,
+        type=helpers.check_dir_arg,
+        help="Path to a directory with testing artifacts",
+    )
+    return parser.parse_args()
+
+
+def main() -> int:
+    logging.basicConfig(
+        format="%(name)s:%(levelname)s:%(message)s",
+        level=logging.INFO,
+    )
+    args = get_args()
+
+    socket_env = os.environ.get("CARDANO_NODE_SOCKET_PATH")
+    if not socket_env:
+        LOGGER.error("The `CARDANO_NODE_SOCKET_PATH` environment variable is not set.")
+        return 1
+
+    state_dir = pl.Path(socket_env).parent
+    cluster_obj = clusterlib.ClusterLib(state_dir=state_dir)
+    location = pl.Path(args.artifacts_base_dir).expanduser().resolve()
+    balance, rewards = testnet_cleanup.addresses_info(cluster_obj=cluster_obj, location=location)
+    LOGGER.info(f"Uncleaned balance: {balance}")
+    LOGGER.info(f"Uncleaned rewards: {rewards}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

cardano_node_tests/utils/testnet_cleanup.py

@@ -381,6 +381,43 @@ def _get_faucet_payment_rec(
     return faucet_payment
 
 
+def addresses_info(cluster_obj: clusterlib.ClusterLib, location: pl.Path) -> tp.Tuple[int, int]:
+    """Return the total balance and rewards of all addresses in the given location."""
+    balance = 0
+    rewards = 0
+    files_found = group_addr_files(find_addr_files(location))
+
+    for files in files_found:
+        for fpath in files:
+            f_name = fpath.name
+            if f_name == "faucet.addr":
+                continue
+
+            # Add sleep to prevent
+            # "Network.Socket.connect: <socket: 11>: resource exhausted"
+            time.sleep(0.1)
+
+            if f_name.endswith("_stake.addr"):
+                address = clusterlib.read_address_from_file(fpath)
+                stake_addr_info = cluster_obj.g_query.get_stake_addr_info(address)
+                if not stake_addr_info:
+                    continue
+                f_rewards = stake_addr_info.reward_account_balance
+                if f_rewards:
+                    rewards += f_rewards
+                    LOGGER.info(f"{f_rewards} on '{fpath}'")
+            else:
+                address = clusterlib.read_address_from_file(fpath)
+                f_balance = cluster_obj.g_query.get_address_balance(
+                    address=address, coin=clusterlib.DEFAULT_COIN
+                )
+                if f_balance:
+                    LOGGER.info(f"{f_balance} on '{fpath}'")
+                    balance += f_balance
+
+    return balance, rewards
+
+
 def cleanup(
     cluster_obj: clusterlib.ClusterLib,
     location: clusterlib.FileType,