Tcp mockserver

Testsuite provides facility to implement custom TCP/IP mockserver. It uses asyncio.StreamReaderProtocol undercover, so you can write simple server that interacts with (reader, writer). Just like with regular asyncio.loop.create_server().

Example

import contextlib

import pytest


@pytest.fixture(scope='session')
async def _tcp_mockserver(create_tcp_mockserver):
    """
    Returns base per-session server instance bound to random port.
    """
    async with create_tcp_mockserver(host='localhost', port=0) as mockserver:
        yield mockserver


@pytest.fixture
def tcp_mockserver(_tcp_mockserver):
    """
    Returns per-test mockserver interface.
    """

    async def handle_client(reader, writer):
        writer.write(b'Hello, world!')
        await writer.drain()
        writer.close()

    with _tcp_mockserver.client_handler(handle_client):
        yield


@pytest.fixture
async def tcp_mockserver_connect(_tcp_mockserver):
    """Create connection to the tcp mockserver."""
    return _tcp_mockserver.open_connection


async def test_server(tcp_mockserver, tcp_mockserver_connect):
    async with tcp_mockserver_connect() as (reader, _):
        data = await reader.read()
        assert data == b'Hello, world!'

Fixtures

create_tcp_mockserver

testsuite.plugins.tcp_mockserver.create_tcp_mockserver(*, host='localhost', port=0, sock=None, **kwargs) → Mockserver

Parameters:

• host – hostname to bind to, default is localhost

• port – port to bind to, default 0 binds to random port

• sock – socket to use instead of (hostname, port) pair

• kwargs – extra params are passed to asyncio.loop.create_server()

Returns an instance of Mockserver.

async with create_tcp_mockserver(host='localhost', port=0) as mockserver:
    yield mockserver

Classes

class testsuite.plugins.tcp_mockserver.Mockserver(server)

TCP/IP mockserver.

property address: Tuple[str, int]

Returns service address (host, port)

with client_handler(handler)

Context manager to install per-test client handler.

async def handle_client(reader, writer):
    writer.write(b'hello\r\n')
    await writer.drain()
    writer.close()

with _tcp_mockserver.client_handler(handle_client):
    ...

async with open_connection(timeout=10.0)

Async context manager creates connection to the service.

Parameters:

timeout – timeout to establish connection.

Returns pair (read, writer).

Connection is closed when context manager is done.

Wrapper around asyncio.open_connection()

async with server.open_connection() as (reader, writer):
    ...

property sockets: Tuple[socket]

Returns list of server sockets.