Chat backend example

Task of storing messages is delegated to an external service

Chat backend service

import argparse
import pathlib

import aiohttp
from aiohttp import web

HTTP_TIMEOUT = 10


def main():
    parser = argparse.ArgumentParser(
        description='Testsuite service integration example.',
    )
    parser.add_argument(
        '--storage-service-url',
        help='Url of service responsible for storing messages',
    )
    parser.add_argument('--port', type=int, default=8080)
    args = parser.parse_args()
    storage_service_url = args.storage_service_url
    routes = web.RouteTableDef()
    root = pathlib.Path(__file__).parent
    routes.static('/static', root.joinpath('static'))

    @routes.get('/ping')
    async def handle_ping(request):
        return web.Response(text='OK.')

    @routes.post('/messages/retrieve')
    async def handle_list(request):
        async with aiohttp.ClientSession() as session:
            response = await session.post(
                storage_service_url + 'messages/retrieve',
                timeout=HTTP_TIMEOUT,
            )
            response.raise_for_status()
            response_body = await response.json()
            messages = list(reversed(response_body['messages']))
            result = {'messages': messages}
        return web.json_response(result)

    @routes.post('/messages/send')
    async def handle_send(request):
        body = await request.json()
        message = {'username': body['username'], 'text': body['text']}
        async with aiohttp.ClientSession() as session:
            response = await session.post(
                storage_service_url + 'messages/send',
                json=message,
                timeout=HTTP_TIMEOUT,
            )
            response.raise_for_status()
        return web.Response(status=204)

    @routes.get('/')
    def handle_root(request):
        return web.FileResponse(root.joinpath('chat.html'))

    app = web.Application()
    app.add_routes(routes)
    web.run_app(app, port=args.port)


if __name__ == '__main__':
    main()

Conftest

import pathlib
import sys

import pytest

pytest_plugins = ['testsuite.pytest_plugin']


def pytest_addoption(parser):
    group = parser.getgroup('Example service')
    group.addoption(
        '--example-service-port',
        help='Bind example services to this port (default is %(default)s)',
        default=8080,
        type=int,
    )


@pytest.fixture
async def example_service(
    ensure_daemon_started,
    # Service process holder
    example_service_scope,
    # Service dependencies
    mockserver,
):
    # Start service if not started yet
    await ensure_daemon_started(example_service_scope)


@pytest.fixture
async def example_client(
    create_service_client,
    example_service_baseurl,
    example_service,
):
    # Create service client instance
    return create_service_client(example_service_baseurl)


@pytest.fixture(scope='session')
def example_service_baseurl(pytestconfig):
    return f'http://localhost:{pytestconfig.option.example_service_port}/'


@pytest.fixture(scope='session')
def example_root():
    """Path to example service root."""
    return pathlib.Path(__file__).parent.parent


@pytest.fixture(scope='session')
async def example_service_scope(
    pytestconfig,
    create_daemon_scope,
    mockserver_info,
    example_root,
    example_service_baseurl,
):
    async with create_daemon_scope(
        args=[
            sys.executable,
            str(example_root.joinpath('server.py')),
            '--port',
            str(pytestconfig.option.example_service_port),
            '--storage-service-url',
            mockserver_info.url('storage/'),
        ],
        ping_url=example_service_baseurl + 'ping',
    ) as scope:
        yield scope

Test

async def test_messages_send(example_client, mockserver):
    @mockserver.handler('/storage/messages/send')
    async def handle_send(request):
        assert request.json == {
            'username': 'Bob',
            'text': 'Hello, my name is Bob!',
        }
        return mockserver.make_response(status=204)

    response = await example_client.post(
        'messages/send',
        json={'username': 'Bob', 'text': 'Hello, my name is Bob!'},
    )
    assert response.status == 204
    assert handle_send.times_called == 1


async def test_messages_retrieve(example_client, mockserver):
    messages = [
        {
            'username': 'Bob',
            'created': '2020-01-01T12:01:00.000',
            'text': 'Hi, my name is Bob!',
        },
        {
            'username': 'Alice',
            'created': {'$date': '2020-01-01T12:02:00.000'},
            'text': 'Hi Bob!',
        },
    ]

    @mockserver.json_handler('/storage/messages/retrieve')
    async def handle_retrieve(request):
        return {'messages': messages}

    response = await example_client.post('messages/retrieve')
    assert response.status == 200
    body = response.json()
    assert body == {'messages': list(reversed(messages))}
    assert handle_retrieve.times_called == 1