Pytest cross suite websocket session-CodePudding

I'm designing an automated test suite to simulate a client which logs in to the backend via api rest and then opens up a websocket communication. I have to test different features over REST and Websocket. Currently I'm performing each websocket test like this:

-The client logs in -The ws communication starts -It sends a ws message and awaits a response -It checks the respose schema and asserts the result -The ws connection is closed -The test end

My problem:

When I run multiple websocket tests as I described above, they open and close the websocket communication several times and my test client ends up being "Blacklisted" because of the irregular behaveour, and from there is not able to reconnect via ws for a considerable time period.

My question: How do I open a websocket connection and keep it open and active across all my tests?

I'm using pytest with "requests" module for api calls and "websockets" module for ws communication.

I've tried to split the python process into two sub-processes with "multiprosessig" module but I'm quite lost here because I'm not able yet to commuicate the pytest process with the websocket process to send the messages and retrive the responses

My websocket connection logic is the following code:

async def websocket_connection(device: Device, cmd_list: list[WebsocketMsg] = None):
    init_cmd = WsInitCommand(device)
    cmd_list.insert(0, init_cmd)

    async def wait_for_correct_response(ws_connection, msj_id: str) -> dict:
        response_received = False
        ws_response: dict = {}
        while not response_received:
            ws_response = json.loads(await ws_connection.recv())
            if 'id' in ws_response and ws_response['id'] == msj_id:
                response_received = True
        return ws_response


    async with websockets.connect(init_cmd.url, subprotocols=init_cmd.sub_protocols) as websocket:
        for cmd in cmd_list:
            await websocket.send(str(cmd.message))
            msg_response: dict = await wait_for_correct_response(websocket, cmd.msg_id)
        return True

CodePudding user response：

Use pytest fixture on a session scope to share singleton websocket connection across tests https://docs.pytest.org/en/stable/reference/fixtures.html#higher-scoped-fixtures-are-executed-first.

Omit multiprocesses and splitting into two processes as it would bring additional complexity and will be tricky to implement.

UPD: answering the comment with this example

import pytest

class WS:
    pass

@pytest.fixture(scope="session")
def websocket_conn():
    ws = WS()
    # establish connection
    print("This is the same object for each test", id(ws))
    yield ws
    # do some cleanup, close connection
    del ws

def test_something(websocket_conn):
    print("Receiving data with", websocket_conn)
    websocket_conn.state = "modified"
    print("Still the same", id(websocket_conn))

def test_something_else(websocket_conn):
    print("Sending data with", websocket_conn)
    print("State preserved:", websocket_conn.state)
    print("Still the same", id(websocket_conn))

tests/test_ws.py::test_something This is the same object for each test 4498209520
Receiving data with <tests.test_ws.WS object at 0x10c1d3af0>
Still the same 4498209520
PASSED
tests/test_ws.py::test_something_else Sending data with <tests.test_ws.WS object at 0x10c1d3af0>
State preserved: modified
Still the same 4498209520
PASSED