Initiating opcua-asyncio Client in thread

Question

In context of writing tests for an implementation using asyncua.Client, I'd like to (1) write a fixture for asyncua.Server and (2) call the function that eventually starts the client in a separate thread. A minimal example of what I'd like to achieve, without the complexity of writing conftest.py etc, is running something like below:

import asyncio
import asyncua

import threading

async def listen(endpoint):
    async with asyncua.Client(url=endpoint) as client:
        print(f'OPC UA client has started {client}.')
        return

async def run_all(endpoint):
    server = asyncua.Server()
    await server.init()
    server.set_endpoint(endpoint)
    async with server:
        t = threading.Thread(target=asyncio.run, args=(listen(endpoint),))
        t.start()
        t.join()

if __name__ == '__main__':
    asyncio.run(run_all('opc.tcp://0.0.0.0:5840/freeopcua/server/'))

This always leads to a TimeoutError in the line async with asyncua.Client.

Most examples I've found have two files called server.py for initializing asyncua.Server and client.py for initializing asyncua.Client. Then the server and client are started in separate terminals. However in order to run pytest, I believe both should start from the same entrypoint. How can this be achieved?

Answer 1

You should not mix async und threading, for such use cases. Use Tasks:

import asyncio
import asyncua


async def listen(endpoint):
    async with asyncua.Client(url=endpoint) as client:
        print(f'OPC UA client has started {client}.')
        return

async def run_all(endpoint):
    server = asyncua.Server()
    await server.init()
    server.set_endpoint(endpoint)
    async with server:
        task = asyncio.create_task(listen(endpoint))
        await task

if __name__ == '__main__':
    asyncio.run(run_all('opc.tcp://0.0.0.0:5840/freeopcua/server/'))