Asynchronously unit testing a Sanic app throws RuntimeError: this event loop is already running

Question

I have a Sanic app that makes some async calls to an external api. I wish to write some unit tests which mock these external calls.

In the code below the tests do pass as we can see from the logs. However after they have completed a RuntimeError: this event loop is already running is thrown

Simplified Sanic app:

app = Sanic(__name__)
app.config.from_pyfile('/usr/src/app/config.py')
Initialize(
    app,
    access_token_name='jwt',
    authenticate=lambda: True,
    claim_aud=app.config.AUTH_JWT_TOKEN['service']['audience'],
    claim_iss=app.config.AUTH_JWT_TOKEN['service']['issuer'],
    public_key=app.config.AUTH_JWT_TOKEN['service']['secret'],
    responses_class=JWTResponses
)


@app.listener('before_server_start')
def init(app, loop):
    ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ssl_ctx.load_cert_chain(app.config.SSL_CERT, app.config.SSL_CERT_KEY)
    ssl_ctx.load_verify_locations(app.config.SSL_SERVER_CERT)
    ssl_ctx.check_hostname = False
    ssl_ctx.verify_mode = ssl.CERT_REQUIRED
    conn = aiohttp.TCPConnector(ssl_context=ssl_ctx)
    app.aiohttp_session = aiohttp.ClientSession(loop=loop, connector=conn)
    access_logger.disabled = True


@app.listener('after_server_stop')
def finish(app, loop):
    loop.run_until_complete(app.aiohttp_session.close())
    loop.close()


@app.route("endpoint/<mpn>")
@protected()
async def endpoint(request, mpn):
    msg = msg(
        mpn,
    )
    headers = {'content-type': 'text/xml'}
    async with session.post(
        config.URL,
        data=msg.tostring(pretty_print=True, encoding='utf-8'),
        headers=headers,
    ) as response:
        response_text = await response.text()
        try:
            response = (
                Response.from_xml(response_text)
            )
            return response
        except ResponseException:
            logger.error(e.get_message()['errors'][0]['message'])
            return response.json(
                e.get_message(),
                status=HTTPStatus.INTERNAL_SERVER_ERROR
            )


if __name__ == '__main__':
    app.run(host="0.0.0.0", port=8000)

And here is the test:

from server import app as sanic_app


@pytest.yield_fixture
def app():
    app = sanic_app
    yield app


@pytest.fixture
def test_cli(loop, app, sanic_client):
    return loop.run_until_complete(sanic_client(app))


token = jwt.encode(
    {
        "iss": (
            sanic_app.config.AUTH_JWT_TOKEN['service']
            ['issuer']
        ),
        "aud": (
            sanic_app.config.AUTH_JWT_TOKEN['service']
            ['audience']
        ),
        "exp": datetime.datetime.utcnow() + datetime.timedelta(
            seconds=int(100)
        )
    },
    sanic_app.config.AUTH_JWT_TOKEN['service']['secret'],
    algorithm='HS256'
).decode('utf-8')
token = 'Bearer ' + token


async def test_success(test_cli):
    with aioresponses(passthrough=['http://127.0.0.1:']) as m:
        with open('tests/data/summary.xml') as f:
            data = f.read()
        m.post(
            'https://external_api',
            status=200,
            body=data
        )
        resp = await test_cli.get(
            'endpoint/07000000000',
            headers={"Authorization": token}
        )
        assert resp.status == 200
        resp_json = await resp.json()
        assert resp_json == {SOME_JSON}

As mentioned above the test does pass but then the error is thrown.

================================================================================================= ERRORS ==================================================================================================
____________________________________________________________________________________ ERROR at teardown of test_success ____________________________________________________________________________________

tp = <class 'RuntimeError'>, value = None, tb = None

    def reraise(tp, value, tb=None):
        try:
            if value is None:
                value = tp()
            if value.__traceback__ is not tb:
                raise value.with_traceback(tb)
>           raise value
/usr/local/lib/python3.6/site-packages/six.py:693:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/local/lib/python3.6/site-packages/six.py:693: in reraise
    raise value
/usr/local/lib/python3.6/site-packages/six.py:693: in reraise
    raise value
/usr/local/lib/python3.6/site-packages/pytest_sanic/plugin.py:212: in sanic_client
    loop.run_until_complete(client.close())
uvloop/loop.pyx:1451: in uvloop.loop.Loop.run_until_complete
    ???
/usr/local/lib/python3.6/site-packages/pytest_sanic/utils.py:230: in close
    await self._server.close()
/usr/local/lib/python3.6/site-packages/pytest_sanic/utils.py:134: in close
    await trigger_events(self.after_server_stop, self.loop)
/usr/local/lib/python3.6/site-packages/pytest_sanic/utils.py:25: in trigger_events
    result = event(loop)
server.py:84: in finish
    loop.run_until_complete(app.aiohttp_session.close())
uvloop/loop.pyx:1445: in uvloop.loop.Loop.run_until_complete
    ???
uvloop/loop.pyx:1438: in uvloop.loop.Loop.run_until_complete
    ???
uvloop/loop.pyx:1347: in uvloop.loop.Loop.run_forever
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   ???
E   RuntimeError: this event loop is already running.

uvloop/loop.pyx:448: RuntimeError

Any help or suggestions is greatly appreciated. Thanks in advance

Answer 1

You can also unregister the call to after_server_stop on your test_cli:

test_cli.server.after_server_stop = []