Make SSH session generator support generating session files for all devices, like fanout switches.

ansible/devutil/device_inventory.py

@@ -0,0 +1,93 @@
+import os
+import csv
+import glob
+from typing import Dict, List, Optional
+
+
+class DeviceInfo(object):
+    """Device information."""
+
+    def __init__(
+        self,
+        hostname: str,
+        management_ip: str,
+        hw_sku: str,
+        device_type: str,
+        protocol: str = "",
+        os: str = "",
+    ):
+        self.hostname = hostname
+        self.management_ip = management_ip
+        self.hw_sku = hw_sku
+        self.device_type = device_type
+        self.protocol = protocol
+        self.os = os
+
+    @staticmethod
+    def from_csv_row(row: List[str]) -> "DeviceInfo":
+        # The device CSV file has the following columns (the last 2 are optional):
+        #
+        #   Hostname,ManagementIp,HwSku,Type,Protocol,Os
+        #
+        return DeviceInfo(
+            row[0],
+            row[1].split("/")[0],
+            row[2],
+            row[3],
+            row[4] if len(row) > 4 else "",
+            row[5] if len(row) > 5 else "",
+        )
+
+    def is_ssh_supported(self) -> bool:
+        if self.device_type == "ConsoleServer":
+            return False
+
+        if self.protocol == "snmp":
+            return False
+
+        return True
+
+
+class DeviceInventory(object):
+    """Device inventory from csv files."""
+
+    def __init__(
+        self, inv_name: str, device_file_name: str, devices: Dict[str, DeviceInfo]
+    ):
+        self.inv_name = inv_name
+        self.device_file_name = device_file_name
+        self.devices = devices
+
+    @staticmethod
+    def from_device_files(device_file_pattern: str) -> "List[DeviceInventory]":
+        inv: List[DeviceInventory] = []
+        for file_path in glob.glob(device_file_pattern):
+            device_inventory = DeviceInventory.from_device_file(file_path)
+            inv.append(device_inventory)
+
+        return inv
+
+    @staticmethod
+    def from_device_file(file_path: str) -> "DeviceInventory":
+        print(f"Loading device inventory: {file_path}")
+
+        # Parse inv name from the file path.
+        # The inv name can be deducted from the file name part in the path using format sonic_<inv_name>_devices.csv
+        inv_name = os.path.basename(file_path).split("_")[1]
+
+        devices: Dict[str, DeviceInfo] = {}
+        with open(file_path, newline="") as file:
+            reader = csv.reader(file)
+
+            # Skip the header line
+            next(reader)
+
+            for row in reader:
+                if row:
+                    device_info = DeviceInfo.from_csv_row(row)
+                    devices[device_info.hostname] = device_info
+
+            return DeviceInventory(inv_name, file_path, devices)
+
+    def get_device(self, hostname: str) -> Optional[DeviceInfo]:
+        return self.devices.get(hostname)

ansible/devutil/testbed.py

@@ -5,19 +5,27 @@
 import os
 import re
 import yaml
+from typing import Any, Dict, List, Optional
+
+from devutil.device_inventory import DeviceInfo, DeviceInventory
 
 
 class TestBed(object):
     """Data model that represents a testbed object."""
 
     @classmethod
-    def from_file(cls, testbed_file="testbed.yaml", testbed_pattern=None, hosts=None):
+    def from_file(
+        cls,
+        device_inventories: List[DeviceInventory],
+        testbed_file: str = "testbed.yaml",
+        testbed_pattern: Optional[str] = None,
+    ) -> Dict[str, "TestBed"]:
         """Load all testbed objects from YAML file.
 
         Args:
             testbed_file (str): Path to testbed file.
             testbed_pattern (str): Regex pattern to filter testbeds.
-            hosts (AnsibleHosts): AnsibleHosts object that contains all hosts in the testbed.
+            hosts (HostManager): AnsibleHosts object that contains all hosts in the testbed.
 
         Returns:
             dict: Testbed name to testbed object mapping.
@@ -39,11 +47,11 @@ def from_file(cls, testbed_file="testbed.yaml", testbed_pattern=None, hosts=None
         for raw_testbed in raw_testbeds:
             if testbed_pattern and not testbed_pattern.match(raw_testbed["conf-name"]):
                 continue
-            testbeds[raw_testbed["conf-name"]] = cls(raw_testbed, hosts=hosts)
+            testbeds[raw_testbed["conf-name"]] = cls(raw_testbed, device_inventories)
 
         return testbeds
 
-    def __init__(self, raw_dict, hosts=None):
+    def __init__(self, raw_dict: Any, device_inventories: List[DeviceInventory]):
         """Initialize a testbed object.
 
         Args:
@@ -55,46 +63,26 @@ def __init__(self, raw_dict, hosts=None):
             setattr(self, key.replace("-", "_"), value)
 
         # Create a PTF node object
-        self.ptf_node = TestBedNode(self.ptf, hosts)
-
-        # Loop through each DUT in the testbed and create TestBedNode object
+        self.ptf_node = DeviceInfo(
+            hostname=self.ptf,
+            management_ip=self.ptf_ip.split("/")[0],
+            hw_sku="Container",
+            device_type="PTF",
+            protocol="ssh",
+        )
+
+        # Loop through each DUT in the testbed and find the device info
         self.dut_nodes = {}
         for dut in raw_dict["dut"]:
-            self.dut_nodes[dut] = TestBedNode(dut, hosts)
+            for inv in device_inventories:
+                device = inv.get_device(dut)
+                if device is not None:
+                    self.dut_nodes[dut] = device
+                    break
+            else:
+                print(f"Error: Failed to find device info for DUT {dut}")
 
         # Some testbeds are dummy ones and doesn't have inv_name specified,
         # so we need to use "unknown" as inv_name instead.
         if not hasattr(self, "inv_name"):
             self.inv_name = "unknown"
-
-
-class TestBedNode(object):
-    """Data model that represents a testbed node object."""
-
-    def __init__(self, name, hosts=None):
-        """Initialize a testbed node object.
-
-        Args:
-            name (str): Node name.
-            ansible_vars (dict): Ansible variables of the node.
-        """
-        self.name = name
-        self.ssh_ip = None
-        self.ssh_user = None
-        self.ssh_pass = None
-
-        if hosts:
-            try:
-                host_vars = hosts.get_host_vars(self.name)
-                self.ssh_ip = host_vars["ansible_host"]
-                self.ssh_user = host_vars["creds"]["username"]
-                self.ssh_pass = host_vars["creds"]["password"][0]
-            except Exception as e:
-                print(
-                    "Error: Failed to get host vars for {}: {}".format(
-                        self.name, str(e)
-                    )
-                )
-                self.ssh_ip = None
-                self.ssh_user = None
-                self.ssh_pass = None