CODEWITHSUNDEEP
pythonpytesttelnetlibtelnetlib3
Alex
Alex

Reputation: 44325

How to use telnetlib3 as a fixture as part of a py.test test case?

As telnetlib seems to get deprecated in future python version I am trying to use telnetlib3 instead (using python 3.10.11, windows 10). I want this to use as a fixture for user-friendly py.test tests. So for that I define a class for the telnet3 as follows:

import telnetlib3
import asyncio

class Telnet3:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        # await telnet3.connect() # ???
        # asyncio.run(await self.connect()) # ???

    async def connect(self):
        self.reader, self.writer = await telnetlib3.open_connection(self.host, self.port)

    async def write_read(self, command):
        self.writer.write(command)

        data = await asyncio.wait_for(self.reader.read(4096), timeout=2)
        return data

And I create a fixture in conftest.py as follows:

from xxx import Telnet3

@pytest.fixture
def telnet3_session(config):
    telnet3 = Telnet3(config["HOST"], config["PORT"])
    # await telnet3.connect() # ???
    return telnet3

And then use it in a test case

def test_telnet(telnet3_session):
    telnet3_session.write_read("$SYS,INFO")

here I get the error

RuntimeWarning: coroutine 'Telnet3.write_read' was never awaited

and with

def test_telnet(telnet3_session):
    await telnet3_session.write_read("$SYS,INFO")

I get the error

 SyntaxError: 'await' outside async function

I run the test case as

python -m pytest -s path/to/case.py

So how to handle this case in a way that a non-expert in asyncio (like me) can easily understand and maintain the test case? Maybe there is an alternative to telnetlib3?

Upvotes: 0

Views: 194

Answers (1)

Alex
Alex

Reputation: 44325

In order to use telnetlib3 as part of py.test fixtures you first need to install pytest-asyncio:

python -m pip install pytest-asyncio

Then your Telnet module looks like

import telnetlib3
import asyncio

class Telnet3:
    def __init__(self, host, port):
        """Initialize the class with the host and port."""
        self.host = host
        self.port = port

    async def connect(self):
        """Defines the connect coroutine and returns the telnet welcome message."""
        self.reader, self.writer = await telnetlib3.open_connection(self.host, self.port)
        return await asyncio.wait_for(self.reader.read(4096), timeout=2)

    def write(self, command):
        """Writes a command to the telnet connection."""
        self.writer.write(command + "\r\n")

    async def read(self, expected="prompt >>"):
        """Read telnet data until expected command prompt appears."""
        reply = ""
        while True:
            data = await self.reader.read(4096)
            if data:
                reply += data
            if expected in reply:
                break
        return reply
    
    async def write_read(self, command, timeout=2):
        """Write to telnet and wait for a response"""
        self.write(command)
        try:
            return await asyncio.wait_for(self.read(), timeout=timeout)
        except asyncio.exceptions.TimeoutError as e:
            print("TimeoutError while reading from telnet!")
            return None

Then in conftest.py you create the fixture e.g. as follows:

import pytest_asyncio

@pytest_asyncio.fixture()
async def telnet3_session(config):
    telnet3 = Telnet3(config["HOST"], config["PORT"])
    reply = await telnet3.connect()    
    yield telnet3

And then finally in the test you can use it as follows

@pytest.mark.asyncio
async def test_telnet(telnet3_session):
    reply = await telnet3_session.write_read("YOUR_COMMAND")
    print(reply)

and then it might work! Not sure if this is the most efficient/simple way to implement it, but it seems to work.

Upvotes: 0

Related Questions