pytest-asyncio aiohttp client testing with fake server

test_client_fake_server.py

import asyncio
import pathlib
import socket
import ssl

import pytest

import aiohttp
from aiohttp import web
from aiohttp.resolver import DefaultResolver
from aiohttp.test_utils import unused_port


class FakeResolver:
    _LOCAL_HOST = {0: '127.0.0.1',
                   socket.AF_INET: '127.0.0.1',
                   socket.AF_INET6: '::1'}

    def __init__(self, fakes, *, loop):
        """fakes -- dns -> port dict"""
        self._fakes = fakes
        self._resolver = DefaultResolver(loop=loop)

    async def resolve(self, host, port=0, family=socket.AF_INET):
        fake_port = self._fakes.get(host)
        if fake_port is not None:
            return [{'hostname': host,
                     'host': self._LOCAL_HOST[family], 'port': fake_port,
                     'family': family, 'proto': 0,
                     'flags': socket.AI_NUMERICHOST}]
        else:
            return await self._resolver.resolve(host, port, family)


ON_ME = {
    "name": "John Doe",
    "id": "12345678901234567"
}


ON_MY_FRIENDS = {
    "data": [
        {
            "name": "Bill Doe",
            "id": "233242342342"
        },
        {
            "name": "Mary Doe",
            "id": "2342342343222"
        },
        {
            "name": "Alex Smith",
            "id": "234234234344"
        },
    ],
    "paging": {
        "cursors": {
            "before": "QVFIUjRtc2c5NEl0ajN",
            "after": "QVFIUlpFQWM0TmVuaDRad0dt",
        },
        "next": ("https://graph.facebook.com/v2.7/12345678901234567/"
                 "friends?access_token=EAACEdEose0cB")
    },
    "summary": {
        "total_count": 3
    }
}


class FakeFacebook:

    def __init__(self, *, loop):
        self.loop = loop
        self.app = web.Application(loop=loop)
        self.app.router.add_routes(
            [web.get('/v2.7/me', self.on_me),
             web.get('/v2.7/me/friends', self.on_my_friends)])
        self.handler = None
        self.server = None
        here = pathlib.Path(__file__)
        ssl_cert = here.parent / 'server.crt'
        ssl_key = here.parent / 'server.key'
        self.ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
        self.ssl_context.load_cert_chain(str(ssl_cert), str(ssl_key))

    async def start(self):
        port = unused_port()
        self.handler = self.app.make_handler()
        self.server = await self.loop.create_server(self.handler,
                                                    '127.0.0.1', port,
                                                    ssl=self.ssl_context
                                                    )
        return {'graph.facebook.com': port}

    async def stop(self):
        self.server.close()
        await self.server.wait_closed()
        await self.app.shutdown()
        await self.handler.shutdown()
        await self.app.cleanup()

    async def on_me(self, request):
        return web.json_response(ON_ME)

    async def on_my_friends(self, request):
        return web.json_response(ON_MY_FRIENDS)


@pytest.fixture()
async def info(event_loop):
    fake_facebook = FakeFacebook(loop=event_loop)
    info = await fake_facebook.start()
    yield info
    await fake_facebook.stop()


@pytest.fixture()
async def session(event_loop, info):
    resolver = FakeResolver(info, loop=event_loop)
    connector = aiohttp.TCPConnector(loop=event_loop, resolver=resolver, verify_ssl=False)
    session = aiohttp.ClientSession(connector=connector, loop=event_loop)
    yield session
    await session.close()


@pytest.mark.asyncio
async def test_server_interaction(session):
    token = "ER34gsSGGS34XCBKd7u"
    async with session.get('https://graph.facebook.com/v2.7/me',
                           params={'access_token': token}) as resp:
        assert await resp.json() == ON_ME

    async with session.get('https://graph.facebook.com/v2.7/me/friends',
                           params={'access_token': token}) as resp:
        assert await resp.json() == ON_MY_FRIENDS