Initiating opcua-asyncio Client in thread

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?

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/'))

Besides the approved answer provided by Schroeder, an alternative using a ThreadPoolExecutor that worked in my case (where there is substantial multithreaded code):

import asyncio
import asyncua

import multiprocessing
import concurrent.futures

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

def listen_sync(endpoint, condition):
    return asyncio.run(listen(endpoint, condition))

async def run_server(endpoint, condition):
    server = asyncua.Server()
    await server.init()
    server.set_endpoint(endpoint)
    async with server:
        with condition:
            condition.notify_all()
        while True:
            await asyncio.sleep(1)

def run_server_sync(endpoint, condition):
    return asyncio.run(run_server(endpoint, condition))

async def run_all(endpoint):
    loop = asyncio.get_running_loop()
    condition = multiprocessing.Condition()
    with concurrent.futures.ThreadPoolExecutor(2) as pool:
        server_task = loop.run_in_executor(pool, run_server_sync, endpoint, condition)
        client_task = loop.run_in_executor(pool, listen_sync, endpoint, condition)
        await asyncio.gather(*[server_task, client_task])

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

Relevant references:

• Tests of the opcua-asyncio library itself, as suggested by Schroeder.
• https://stackoverflow.com/questions/69741177/run-multiple-async-loops-in-separate-processes-within-a-main-async-app