Source code for reemote.utilities.verify_inventory_connect
# Copyright (c) 2025 Kim Jarvis TPF Software Services S.A. kim.jarvis@tpfsystems.com # This software is licensed under the MIT License. See the LICENSE file for details.#importasyncsshfromtypingimportList,Tuple,Dict,Any
[docs]asyncdefverify_inventory_connect(inventory:List[Tuple[Dict[str,Any],Dict[str,str]]])->bool:"""Asynchronously verifies SSH connectivity for a list of hosts. This function iterates through an inventory of hosts and attempts to establish an SSH connection to each one using the `asyncssh` library. For every host, it uses a dictionary of connection parameters to initiate the connection. A simple `echo x` command is executed on each successful connection to confirm that the session is active and capable of running commands. If any connection fails, an error is printed to the console, and the function immediately returns `False`, stopping further verification. Args: inventory (List[Tuple[Dict[str, Any], Dict[str, str]]]): A list of tuples, where each tuple represents a host to be verified. Each tuple contains: - host_info (dict): A dictionary of keyword arguments to be passed to `asyncssh.connect`. Common keys include `host`, `username`, `port`, `password`, and `client_keys`. - ssh_info (dict): A dictionary containing other SSH-related metadata (currently unused by this function). Returns: bool: `True` if connections to all hosts in the inventory are successful, and `False` if any connection attempt fails. """forhost_info,ssh_infoininventory:try:# Connect to the hostasyncwithasyncssh.connect(**host_info)asconn:# Run the commandcp=awaitconn.run("echo x")except(OSError,asyncssh.Error)ase:print(f"Connection failed on host {host_info.get("host")}: {str(e)}")returnFalsereturnTrue