7

I have a simple echo server function. Everything works fine if I launch pytest with a single test, but if I launch it with two tests then the second one hangs on waiting for the server to start, and I can't understand why. Here's a complete code.

The server:

async def handle_echo(reader, writer):
    data = await reader.read(100)

    message = data.decode()
    addr = writer.get_extra_info('peername')
    print(f"SERVER: Received {message!r} from {addr!r}")
    writer.write(data)
    await writer.drain()
    print(f"SERVER: Sent: {message!r}")

    writer.close()
    print("SERVER: Closed the connection")

Test setup:

HOST = "localhost"


@pytest.fixture()
def event_loop():
    return asyncio.get_event_loop()


async def _async_wait_for_server(event_loop, addr, port):
    while True:
        a_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        await event_loop.sock_connect(a_socket, (addr, port))
        return
    except ConnectionRefusedError:
        await asyncio.sleep(0.001)

    finally:
        a_socket.close()


@pytest.fixture()
def server(event_loop):
    unused_tcp_port = 65432
    cancel_handle = asyncio.ensure_future(main(unused_tcp_port), loop=event_loop)
    event_loop.run_until_complete(asyncio.wait_for(
        _async_wait_for_server(event_loop, HOST, unused_tcp_port), 5.0))
    try:
        yield unused_tcp_port
    finally:
        cancel_handle.cancel()


async def main(port):
    server = await asyncio.start_server(handle_echo, HOST, port)
    addr = server.sockets[0].getsockname()
    print(f'SERVER: Serving on {addr[0:2]}')
    async with server:
        await server.serve_forever()

Tests:

@pytest.mark.asyncio
async def test_something(server):
    message = "Foobar!"
    reader, writer = await asyncio.open_connection(HOST, server)

    print(f'CLIENT: Sent {message!r}')
    writer.write(message.encode())
    await writer.drain()

    data = await reader.read(100)
    print(f'CLIENT: Received {data.decode()!r}')

    print('CLIENT: Close the connection')
    writer.close()
    await writer.wait_closed()

@pytest.mark.asyncio
async def test_another_thing(server):
    # Literally the same

Pytest output: ================================================================== test session starts =================================================================== platform win32 -- Python 3.7.0, pytest-6.2.5, py-1.10.0, pluggy-1.0.0 -- C:\Users\hurrd.virtualenvs\chatServer-iuEQ-ghN\Scripts\python.exe cachedir: .pytest_cache rootdir: C:\Users\hurrd\PycharmProjects\chatServer plugins: asyncio-0.16.0 collected 2 items

tests/unit/test_themes.py::test_something PASSED [ 50%]

tests/unit/test_themes.py::test_another_thing

1
  • 2
    Same behavior here, still have not figured it out. Commented Jul 2, 2022 at 22:50

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.