При асинхронном модульном тестировании приложения Sanic выдается RuntimeError: этот цикл событий уже запущен - PullRequest
Новая
       51

При асинхронном модульном тестировании приложения Sanic выдается RuntimeError: этот цикл событий уже запущен

1 голос
/ 14 июня 2019

У меня есть приложение Sanic, которое выполняет некоторые асинхронные вызовы для внешнего API.Я хотел бы написать несколько модульных тестов, которые будут имитировать эти внешние вызовы.

В приведенном ниже коде тесты проходят, как мы видим из журналов.Однако после того, как они выполнили RuntimeError: этот цикл событий уже запущен, генерируется

Упрощенное приложение Sanic: 

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)

И вот тест: 

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}

Как упоминалось выше, тест проходит успешно, но затем выдается ошибка.

================================================================================================= 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

Любая помощь или предложения с благодарностью.Заранее спасибо

Ответы [ 2 ]

1 голос
/ 14 июня 2019

Вы также можете отменить регистрацию вызова на after_server_stop на вашем test_cli: 

test_cli.server.after_server_stop = []
0 голосов
/ 14 июня 2019

Я думаю, что вы можете исправить ошибку, изменив этот метод: 

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

Вы не обязаны закрывать цикл, и вы должны использовать тот факт, что finish вызывается из асинхронного контекста (вам не нужно запускать цикл событий, он уже запущен).

Если это не единственная проблема, то, возможно, начните с более простого примера и добавляйте его, пока он снова не сломается. Sanic, похоже, имеет хорошо документированный тестовый раздел , где они предлагают использовать pytest-sanic . Простой пример использования pytest будет выглядеть так: 

# file: server.py
from sanic import Sanic
from sanic.response import json

app = Sanic()

@app.route('/')
async def test(request):
    return json({'hello': 'world'})

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

# file: server_test.py
import pytest
from server import app

@pytest.yield_fixture
def sanic_app():
    yield app

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

async def test_index(test_cli):
    resp = await test_cli.get('/')
    assert resp.status == 200
    json = await resp.json()
    assert json == {'hello': 'world'}

async def test_index_fail(test_cli):
    resp = await test_cli.get('/')
    assert resp.status == 200
    json = await resp.json()
    assert json == {'bonjour': 'monde'}

Вам необходимо установить несколько пакетов: 

pip install sanic pytest pytest-sanic

Тогда вы можете просто запустить pytest, у вас должен быть первый тест пройден, а второй не пройден.

Как правило, вам не нужно запускать цикл событий самостоятельно, всегда пытайтесь избавиться от loop.run_....

...