如何将telnetlib3作为pytest测试用例的fixture并处理异步问题？

先搞定异步测试的基础配置 #

首先得装pytest-asyncio插件，不然pytest没法处理异步用例：

pip install pytest-asyncio

运行测试的时候要么加-p pytest_asyncio参数，要么在项目根目录建个pytest.ini，里面写：

[pytest]
asyncio_mode = auto

改写异步Fixture和测试用例 #

把你的Telnet3类、fixture还有测试用例都改成异步的，就不会报那俩错误了：

conftest.py 示例 #

import pytest
import telnetlib3

class Telnet3:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.reader = None
        self.writer = None

    # 异步连接方法
    async def connect(self):
        self.reader, self.writer = await telnetlib3.open_connection(self.host, self.port)

    # 异步读写方法
    async def write_read(self, command, prompt):
        self.writer.write(f"{command}\n")
        await self.writer.drain()
        output = ""
        while True:
            data = await self.reader.read(1024)
            if not data:
                break
            output += data
            if prompt in output:
                break
        return output

    # 异步关闭连接
    async def close(self):
        if self.writer:
            self.writer.close()
            await self.writer.wait_closed()

# 异步fixture，pytest-asyncio会自动处理事件循环
@pytest.fixture
async def telnet_client():
    client = Telnet3("你的目标主机", 23)
    await client.connect()
    yield client
    await client.close()

测试用例示例 #

# 测试用例必须是async函数，里面用await调用异步方法
async def test_telnet_show_version(telnet_client):
    result = await telnet_client.write_read("show version", "#")
    assert "Version" in result

这样改完，既不会因为没加await报RuntimeWarning，也不会在非异步函数里用await报语法错误——所有异步操作都在async函数里，且都加了await。

不想碰异步？用同步Telnet库 #

要是完全不想搞async那一套，直接换同步的Telnet库就行，推荐两个：

1. netmiko #

专门做网络设备自动化的库，支持Telnet和SSH，API全是同步的，上手简单：

pip install netmiko

示例代码：

from netmiko import TelnetHandler

def test_netmiko_telnet():
    # 用上下文管理器自动处理连接关闭
    with TelnetHandler(host="你的主机", username="用户名", password="密码") as conn:
        output = conn.send_command("show version", expect_string="#")
        assert "Version" in output

2. telnetlibx #

是telnetlib的替代库，完全兼容原有telnetlib的同步API，几乎可以直接替换你原来的代码，不需要改逻辑：

pip install telnetlibx

用法和原来的telnetlib一模一样，比如：

import telnetlibx

def test_telnetlibx():
    tn = telnetlibx.Telnet("你的主机", 23)
    tn.read_until(b"login: ")
    tn.write(b"用户名\n")
    tn.read_until(b"Password: ")
    tn.write(b"密码\n")
    tn.write(b"show version\n")
    output = tn.read_until(b"#").decode()
    assert "Version" in output
    tn.close()

内容的提问来源于stack exchange，提问作者Alex