Add CsonicHost for docker-exec based neighbor access in cSONiC testbeds

tests/common/devices/csonic.py

@@ -0,0 +1,158 @@
+"""
+CsonicHost - A lightweight host class for cSONiC (docker-sonic-vs) neighbor containers.
+
+Instead of SSH/Ansible, this class uses 'docker exec' on the VM host to run commands
+inside the cSONiC container. This avoids the need for sshd, admin user, or mgmt IP
+inside the container.
+"""
+
+import json
+import logging
+import subprocess
+
+from tests.common.devices.base import NeighborDevice
+
+logger = logging.getLogger(__name__)
+
+
+class CsonicHost(object):
+    """
+    A neighbor host running as a cSONiC (docker-sonic-vs) Docker container.
+
+    Provides a command/shell interface compatible with SonicHost/EosHost by
+    executing commands via 'docker exec' on the VM host rather than SSH.
+    """
+
+    def __init__(self, container_name, vm_host_ip=None, vm_host_user=None):
+        """
+        Args:
+            container_name: Docker container name (e.g., 'csonic_vms6-1_VM0100')
+            vm_host_ip: IP of the host running the container (default: localhost)
+            vm_host_user: SSH user for the VM host (only needed if remote)
+        """
+        self.container_name = container_name
+        self.hostname = container_name
+        self.vm_host_ip = vm_host_ip
+        self.vm_host_user = vm_host_user
+        self.is_local = vm_host_ip is None or vm_host_ip in ('localhost', '127.0.0.1')
+
+    def __str__(self):
+        return '<CsonicHost {}>'.format(self.container_name)
+
+    def __repr__(self):
+        return self.__str__()
+
+    def _docker_exec(self, cmd, **kwargs):
+        """Run a command inside the Docker container via docker exec."""
+        docker_cmd = ['docker', 'exec', self.container_name, 'bash', '-c', cmd]
+
+        if not self.is_local:
+            ssh_prefix = ['ssh', '-o', 'StrictHostKeyChecking=no',
+                          '-o', 'UserKnownHostsFile=/dev/null']
+            if self.vm_host_user:
+                ssh_prefix.extend(['-l', self.vm_host_user])
+            ssh_prefix.append(self.vm_host_ip)
+            # Wrap docker command for remote execution
+            docker_cmd = ssh_prefix + [' '.join(
+                "'{}'".format(c) if ' ' in c else c for c in docker_cmd
+            )]
+
+        logger.debug("CsonicHost [%s] executing: %s", self.container_name, cmd)
+
+        try:
+            result = subprocess.run(
+                docker_cmd,
+                capture_output=True,
+                text=True,
+                timeout=kwargs.get('timeout', 30)
+            )
+            stdout = result.stdout.strip()
+            stderr = result.stderr.strip()
+            rc = result.returncode
+
+            response = {
+                'stdout': stdout,
+                'stdout_lines': stdout.split('\n') if stdout else [],
+                'stderr': stderr,
+                'stderr_lines': stderr.split('\n') if stderr else [],
+                'rc': rc,
+                'failed': rc != 0 and not kwargs.get('module_ignore_errors', False),
+            }
+
+            if rc != 0 and not kwargs.get('module_ignore_errors', False):
+                logger.warning("CsonicHost [%s] command failed (rc=%d): %s\nstderr: %s",
+                               self.container_name, rc, cmd, stderr)
+
+            return response
+
+        except subprocess.TimeoutExpired:
+            logger.error("CsonicHost [%s] command timed out: %s", self.container_name, cmd)
+            return {
+                'stdout': '',
+                'stdout_lines': [],
+                'stderr': 'Command timed out',
+                'stderr_lines': ['Command timed out'],
+                'rc': -1,
+                'failed': True,
+            }
+
+    def command(self, cmd, **kwargs):
+        """Run a command (compatible with Ansible command module interface)."""
+        return self._docker_exec(cmd, **kwargs)
+
+    def shell(self, cmd, **kwargs):
+        """Run a shell command (compatible with Ansible shell module interface)."""
+        return self._docker_exec(cmd, **kwargs)
+
+    def shutdown(self, ifname):
+        """Shut down an interface."""
+        logger.info("CsonicHost [%s] shutting down %s", self.container_name, ifname)
+        return self._docker_exec("ip link set {} down".format(ifname))
+
+    def no_shutdown(self, ifname):
+        """Bring up an interface."""
+        logger.info("CsonicHost [%s] bringing up %s", self.container_name, ifname)
+        return self._docker_exec("ip link set {} up".format(ifname))
+
+    def get_route(self, prefix):
+        """Get route info from FRR."""
+        result = self._docker_exec("vtysh -c 'show ip route {} json'".format(prefix))
+        if result['rc'] == 0 and result['stdout']:
+            try:
+                return json.loads(result['stdout'])
+            except json.JSONDecodeError:
+                pass
+        return {}
+
+    def get_port_channel_status(self, pc_name=None):
+        """Get PortChannel status."""
+        if pc_name:
+            result = self._docker_exec("teamdctl {} state dump".format(pc_name))
+        else:
+            result = self._docker_exec("show interfaces portchannel")
+        if result['rc'] == 0 and result['stdout']:
+            try:
+                return json.loads(result['stdout'])
+            except (json.JSONDecodeError, ValueError):
+                return result['stdout']
+        return {}
+
+    def config(self, lines=None, parents=None):
+        """
+        Configure via vtysh (loose compatibility with EOS config style).
+        Translates config lines to vtysh commands.
+        """
+        if not lines:
+            return {}
+        cmds = []
+        if parents:
+            for p in parents:
+                cmds.append(p)
+        for line in lines:
+            cmds.append(line)
+
+        vtysh_cmd = "vtysh"
+        for c in cmds:
+            vtysh_cmd += " -c '{}'".format(c)
+
+        return self._docker_exec(vtysh_cmd)

tests/conftest.py

@@ -27,6 +27,7 @@
 from tests.common.devices.ptf import PTFHost
 from tests.common.devices.eos import EosHost
 from tests.common.devices.sonic import SonicHost
+from tests.common.devices.csonic import CsonicHost
 from tests.common.devices.fanout import FanoutHost
 from tests.common.devices.k8s import K8sMasterHost
 from tests.common.devices.k8s import K8sMasterCluster
@@ -155,7 +156,8 @@ def pytest_addoption(parser):
                      help="Name of k8s master group used in k8s inventory, format: k8s_vms{msetnumber}_{servernumber}")
 
     # neighbor device type
-    parser.addoption("--neighbor_type", action="store", default="eos", type=str, choices=["eos", "sonic", "cisco"],
+    parser.addoption("--neighbor_type", action="store", default="eos", type=str,
+                     choices=["eos", "sonic", "cisco", "csonic"],
                      help="Neighbor devices type")
 
     # ceos neighbor lacp multiplier
@@ -930,6 +932,15 @@ def initial_neighbor(neighbor_name, vm_name, multi_vrf_peer=False, multi_vrf_pri
                     'conf': tbinfo['topo']['properties']['configuration'][neighbor_name]
                 }
             )
+        elif neighbor_type == "csonic":
+            vm_set_name = tbinfo.get('group-name', '')
+            container_name = "csonic_{}_{}".format(vm_set_name, vm_name)
+            device = NeighborDevice(
+                {
+                    'host': CsonicHost(container_name),
+                    'conf': tbinfo['topo']['properties']['configuration'][neighbor_name]
+                }
+            )
         else:
             raise ValueError("Unknown neighbor type %s" % (neighbor_type,))
         devices[neighbor_name] = device
@@ -1351,7 +1362,7 @@ def collect_techsupport_all_duts(request, duthosts):
 @pytest.fixture
 def collect_techsupport_all_nbrs(request, nbrhosts):
     yield
-    if request.config.getoption("neighbor_type") == "sonic":
+    if request.config.getoption("neighbor_type") in ("sonic", "csonic"):
         [collect_techsupport_on_dut(request, nbrhosts[nbrhost]['host']) for nbrhost in nbrhosts]
 
 

tests/vlan/test_vlan_ping.py

@@ -48,7 +48,7 @@ def static_neighbor_entry(duthost, dic, oper, ip_version="both"):
 
 
 @pytest.fixture(scope='module')
-def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo, lower_tor_host):   # noqa: F811
+def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo, lower_tor_host, request):   # noqa: F811
     """
     Setup:      Collecting vm_host_info, ptfhost_info
     Teardown:   Removing all added ipv4 and ipv6 neighbors
@@ -65,17 +65,19 @@ def vlan_ping_setup(duthosts, rand_one_dut_hostname, ptfhost, nbrhosts, tbinfo,
 
     py_assert(vm_name is not None, "Can't get neighbor vm")
 
+    neighbor_type = request.config.getoption("--neighbor_type")
+
     # Determine which interface to use
     if topo_type == "mx":
         interface_name = 'Ethernet1'
         dev_name = 'eth1'
     else:
         if 'Port-Channel1' in vm_info['conf']['interfaces']:
             interface_name = 'Port-Channel1'
-            dev_name = 'po1'
+            dev_name = 'PortChannel1' if neighbor_type == 'csonic' else 'po1'
         else:
             interface_name = 'Ethernet1'
-            dev_name = 'eth1'
+            dev_name = 'Ethernet0' if neighbor_type == 'csonic' else 'eth1'
         # in case of lower tor host we need to use the next portchannel
         if "dualtor-aa" in tbinfo["topo"]["name"] and rand_one_dut_hostname == lower_tor_host.hostname:
             interface_name = 'Port-Channel2'