在一般涉及网络转发的测试的过程，涉及的设备节点往往不止一个，而测试的需要往往需要在多个节点上来完成流量的发送，数据的接收，数据包、连接的计数检查等。如果依赖手动地登录每一台设备并且运行相应地脚本命令的话，那么会削减测试的效率，并且也需要将每台设备上的不通结果根据不同的连接对应起来也会对调试造成一定的困难。而使用netmiko实现远程登录多个节点并且结合pytest来实现自动化测试是解决多网络节点流量测试的一个方案。
主要环境依赖：python3/pytest/netmiko/paramiko

1：基于netmiko实现多节点登录：

netmiko登录主要依赖于ssh协议，登录设备信息往往需要以一个字典的形式来表示。字典中主要包含设备登录ip、设备登录的用户名、密码这样的信息。同时也可以进一步设置超时时间。设备字典结构一般如下：

        self.device = {
            'device_type': 'linux',
            'ip': '',
            'port': 22,
            'username': '',
            'password': '',
            'conn_timeout': 3,
        }

netmiko在接收device字典之后，会根据设备类型device_type来合适的连接类。

def ssh_dispatcher(device_type: str) -> Type["BaseConnection"]:
    """Select the class to be instantiated based on vendor/platform."""
    return CLASS_MAPPER[device_type]

netmiko支持的部分设备类型。这里以linux设备类型为例，device_type设置为linux设备类型之后，会选取LinuxSSH类，不同的类里面封装了对不同设备的处理逻辑。

    CLASS_MAPPER_BASE = {"linux": LinuxSSH,
    "mikrotik_routeros": MikrotikRouterOsSSH,
    "mikrotik_switchos": MikrotikSwitchOsSSH,
    "mellanox": MellanoxMlnxosSSH,
    "mellanox_mlnxos": MellanoxMlnxosSSH,
    "mrv_lx": MrvLxSSH,
    "mrv_optiswitch": MrvOptiswitchSSH,
,}

常用的LinuxSSH类中：封装了很多像check_config_model之类的函数，用于验证是否处于特权模式（enable mode），当然这些函数在使用netmiko来登录到多个节点的时候并不需要关心具体的实现细节。

class LinuxSSH(CiscoSSHConnection):
    prompt_pattern = rf"[{re.escape(LINUX_PROMPT_PRI)}{re.escape(LINUX_PROMPT_ALT)}]"

    def check_config_mode(
        self,
        check_string: str = LINUX_PROMPT_ROOT,
        pattern: str = "",
        force_regex: bool = False,
    ) -> bool:
        """Verify root"""
        return self.check_enable_mode(check_string=check_string)

    def config_mode(
        self,
        config_command: str = "sudo -s",
        pattern: str = "ssword",
        re_flags: int = re.IGNORECASE,
    ) -> str:
        """Attempt to become root."""
        return self.enable(cmd=config_command, pattern=pattern, re_flags=re_flags)

选取合适的连接类之后，根据设备，netmiko的ConnectHandler函数会返回一个连接实例，这个连接实例会用于与指定设备之间的一系列交互，

ConnectHandler函数定义如下，它会根据提供的设备类型来返回对应的连接实例。获取到这个连接实例之后，相当于netmiko已经登录到了这个网络设备，并且可以根据netmiko来对这个设备进行相应的操作。

def ConnectHandler(*args: Any, **kwargs: Any) -> "BaseConnection":
    """Factory function selects the proper class and creates object based on device_type."""
    device_type = kwargs["device_type"]
    if device_type not in platforms:
        if device_type is None:
            msg_str = platforms_str
        else:
            msg_str = telnet_platforms_str if "telnet" in device_type else platforms_str
        raise ValueError(
            "Unsupported 'device_type' "
            "currently supported platforms are: {}".format(msg_str)
        )
    ConnectionClass = ssh_dispatcher(device_type)
    return ConnectionClass(*args, **kwargs)

2：结合pytest实现测试命令的自动分发执行

2.1：以一个client向server端发送请求，server端给client发送reply为例。这里每个设备配备了mgmt_ip和data_ip，mgmt_ip主要是用于netmiko的连接和指令的发送，而data_ip则是用于实际的业务流量测试。具体定义设备的字典如下：

    'client': {
        'port': 22,
        'username': '',
        'password': '',
        'mgmt_ip': '',
        'data_ip': '',
        'data_ip6': '',
        'data_mac': ''
    },

2.2：基于pytest的参数化方法可以指定不同的数据包发送的大小、长度、个数等，并且自动执行，和测试。基于不同的参数组合成一条发送数据包的命令，这里以icmp数据包为例， 根据不同的参数，pytest使得每个测试case中发送不同的icmp数据包。同时在pytest中也有两个特殊的方式setup_class和teardown方法，这两个方法会分别在测试开始和测试结束的时候来执行，对于使用netmiko的场景而言，往往在测试开始的时候会登录所有设备，而在测试结束的时候会退出所有的设备。对于最基本的连通性测试而言，可以用client来向server端来发送不同的icmp数据包，并且根据控制台的输出来判断是否有丢包现象。

    def setup_class(self):
        LOG.info("Setting up env...")
        self.client = Client(topo_conf['client'])
        self.client.login()
        self.server = Server(topo_conf['server'])
        self.server.login()
    def teardown_class(self):
        self.client.logout()
        self.server.logout()
    @pytest.mark.parametrize('pkt_cnt, pkt_size, intval, timeout', [
        (10, 64, 0.1, 2),
        (10, 128, 0.1, 2),
    ])
    def test_icmp(self, pkt_cnt, pkt_size, intval, timeout):
        out = self.client.ping(self.server.data_ip, pkt_cnt, pkt_size, intval, timeout)
        time.sleep(2)
        check_icmp(out)

2.3 基于netmiko分发指令给不同的设备。

登录每个设备的时候会获取到一个netmiko提供的connect实例。根据这个连接，netmiko提供了一些方法来给指定的设备发送命令，并且可以返回对应设备上的输出。这样就可以实现远程地操作指定设备。Send_command为netmiko提供的发送命令的函数，可以通过connect.send_command来进行调用。除了分发命令之外同时，netmiko和paramiko也提供了在不同设备之间实现文件传输的方式。像在paramiko中提供了transfer_file方法，可以实现将本地设备上的文件发送到远程设备上的功能以及将远程设备上的文件发送到本地的功能。主要涉及put和get两个方法。

首先，创建一个 SSHClient 对象 SSH。接下来，根据是否提供了私钥，分别设置 SSH 客户端的连接方式。如果没有提供私钥，使用密码方式连接并设置自动添加主机密钥。如果提供了私钥，使用私钥方式连接并根据私钥是否加密设置密码。然后，通过 SSHClient 对象获取传输对象的transport。利用传输对象创建一个 SFTPClient 对象c_sftp，用于进行文件传输操作。如果get参数为 True，表示需要从远程主机获取文件。以日志的形式记录正在进行的文件传输操作，并调用 SFTPClient 的get方法将远程文件下载到本地目标位置，并将下载的文件对象赋给变量 d。

    def transfer_file(self, src, dest, get=True):
        ssh = paramiko.SSHClient()
        try:
            if self._private_key is None:
                ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                ssh.connect(hostname=self._host, port=self._port, username=self._user, password=self._passwd)
            else:
                if self._private_key_secret is None:
                    key = paramiko.RSAKey.from_private_key_file(self._private_key)
                else:
                    key = paramiko.RSAKey.from_private_key_file(self._private_key, password=self._private_key_secret)

                # ssh.load_system_host_keys()
                ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
                ssh.connect(hostname=self._host, port=self._port, username=self._user, pkey=key)
            t_port = ssh.get_transport()
            c_sftp = paramiko.SFTPClient.from_transport(t_port)
            if get:
                LOG.info(f"getting file {self._host}:{self._port}{src} to {dest}")
                d = c_sftp.get(src, dest)
            else:
                LOG.info(f"putting file {src} to {self._host}:{self._port}{dest}")
                d = c_sftp.put(src, dest)
        except Exception as e:
            LOG.error(f"{e.__class__}: {str(e)}")
            LOG.error(traceback.format_exc())
            raise e
        finally:
            ssh.close()

        return d

2.4：一般测试流程。

以2.1的测试拓扑结构为例，对于网络拓扑中的多个节点，如果用netmiko/paramiko来实现自动化测试主要的流程为：

• Netmiko分别登录到多个设备：client\server等。
• 调用send_command方法向client发送命令，一般为发送数据包的命令。
• 调用send_command方法向拓扑中的其他节点发送命令，一般为抓包命令例如:tcpdump
• 使用paramiko提供的transfer_file方法来将不同节点上抓包的文件发送到代码指定的host设备上，并且分析统计有没有丢包之类的现象。
• 退出设备登录

3. 测试退出及结果检查

3.1：测试退出

测试结束之后首先需要释放netmiko占用的ssh连接，可以调用connect.disconnect方法来退出netmiko的登录。但是这样的退出方法可能会带来较长的时间占用。同样地，netmiko也提供了一个close方法可以直接关闭会话，它会显式地关闭与远程设备的 SSH 连接。它会直接关闭底层的传输层对象，断开与设备的连接。可以通过connect.remote_conn.transport.close()来调用这个函数。对于disconnect方法：

该方法根据连接协议类型进行清理的不同处理。首先尝试调用cleanup方法来关闭会话以及相关的日志和输出流。如果 cleanup发生异常，也不会中断执行。然后，根据设备的连接协议类型执行不同的清理操作。无论清理操作执行是否成功，都会将连接相关的属性设置为None，然后关闭与远程设备的日志记录文件。最后，关闭与设备的会话。如果清理操作发生异常，则会跳过异常直接结束执行。

    def disconnect(self) -> None:
        """Try to gracefully close the session."""
        try:
            self.cleanup()
        except Exception:
            # Keep going on cleanup process even if exceptions
            pass

        try:
            if self.protocol == "ssh":
                self.paramiko_cleanup()
            elif self.protocol == "telnet":
                assert isinstance(self.remote_conn, telnetlib.Telnet)
                self.remote_conn.close()
            elif self.protocol == "serial":
                assert isinstance(self.remote_conn, serial.Serial)
                self.remote_conn.close()
        except Exception:
            # There have been race conditions observed on disconnect.
            pass
        finally:
            self.remote_conn_pre = None
            self.remote_conn = None
            if self.session_log:
                self.session_log.close()

3.2：一般测试报错情况分析：

主要调试可以根据pytest给出的相应的测试报告来调试,对于pytest而言在测试结尾会给出相应的测试报告,测试报告中会给出具体错误的测试用例,以及错误原因,另外也会有具体的错误位置。主要错误可以分为下面几类:

1:assert  error:即断言错误,一般是测试用例的结果不符合预期,可以根据pytest给出的报告定位到具体断言错误的位置,并且调试发现原因。

2: netmiko time error: 即case的错误来源为netmiko,出现这类错误的原因主要可能是,设备断连导致netmiko连接不上或者是流量不通导致netimiko命令发送超时,比如netmiko向client发送了一条ping命令,但是ping命令失败,这会造成netmiko发送命令超时阻塞,这个问题可以通过“ping&"后台执行来解决。

3:干扰流量:在client/server上可以使用tcpdump来抓取数据包，来抓取数据包,并且具体分析数据包,查看时候有无关流量/丢包等情况的发生。