aiohttp_client - RuntimeError: Менеджер контекста времени ожидания должен использоваться внутри задачи - PullRequest
Новая
       92

aiohttp_client - RuntimeError: Менеджер контекста времени ожидания должен использоваться внутри задачи

1 голос
/ 23 февраля 2020

Что я делаю

Я изучаю aiohttp, создавая REST API, который я тестирую с помощью Pytest (и его asyn c и плагинов aiohttp).

Для моего первого теста (с самого начала я буду использовать TDD) у меня есть следующий код: 

@pytest.mark.asyncio
async def test_handle_user_create(
    aiohttp_client, init_test_app, create_test_user_table
):
    payload = {
        "email": "tintin@gmail.com",
        "username": "Tintin",
        "password": "y0u != n00b1e",
    }
    client = await aiohttp_client(init_test_app)
    resp = await client.post("/users/", json=payload)
    ...
  • aiohttp_client - это клиентское устройство от pytest-aiohttp
  • init_test_app - это устройство, которое по существу отражает приложение, которое я собираюсь создать
  • create_test_user_table - это мое устройство для создания таблицы для пользователей в тестовой базе данных

Что с ним не так

Мой первый тест выдает следующую ошибку времени выполнения в последней строке в блоке кода выше: 

_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
env36\lib\site-packages\aiohttp\test_utils.py:295: in request
    method, self.make_url(path), **kwargs
env36\lib\site-packages\aiohttp\client.py:417: in _request
    with timer:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <aiohttp.helpers.TimerContext object at 0x0000015DE23A3B38>

    def __enter__(self) -> BaseTimerContext:
        task = current_task(loop=self._loop)

        if task is None:
>           raise RuntimeError(
                'Timeout context manager should be used '
                'inside a task'
            )
E           RuntimeError: Timeout context manager should
                          be used inside a task

env36\lib\site-packages\aiohttp\helpers.py:568: RuntimeError

From я получаю сообщение об ошибке, что клиент пытается использовать менеджер контекста времени ожидания asyn c, но это не удается, потому что я не вызываю его внутри задачи.

Я не знаю, является ли мой вывод это правильно.

И еще, я не достаточно комфортно с Asyncio, чтобы знать, как решить эту проблему.

* 1 034 * Буду благодарен, если кто-нибудь покажет мне выход.

Подробнее

Вот исходный код моего тестового файла: 

import asyncio
import sqlite3
from pathlib import Path

import pytest
from aiohttp import web

from app import router


@pytest.fixture(name="event_loop", scope="session")
def fixture_event_loop():
    """
    Mock session scoped event loop.

    Default event loop is function scoped, and won't work with
    otherwisely scoped fixtures. Hence, the need for this overwrite.
    """
    loop = asyncio.get_event_loop()
    yield loop
    loop.close()


@pytest.fixture(name="test_db_path", scope="session")
async def fixture_test_db_path():
    return Path(__file__).absolute().parent.joinpath("test_db.sqlite")


@pytest.fixture(name="init_test_db_conn", scope="session")
async def fixture_init_test_db_conn(test_db_path):
    """
    Mock initiator of test database connection.
    """

    async def _init_test_db_conn(test_app):
        with sqlite3.connect(str(test_db_path)) as conn:
            test_app["DB_CONN"] = conn
            yield

    return _init_test_db_conn


@pytest.fixture(name="init_test_app", scope="session")
async def fixture_init_test_app(init_test_db_conn):
    """
    Mock app for testing.

    Substitute the test db for the development db for testing and
    undo the substitution after all tests have been run.
    """
    app = web.Application()
    app.add_routes(router)
    app.cleanup_ctx.append(init_test_db_conn)
    return app


@pytest.fixture(name="create_test_user_table")
def fixture_create_test_user_table(test_db_path):
    """
    Mock user table for tests. Scope at function level.

    Drop table at end of each test.
    """
    conn = sqlite3.connect(str(test_db_path))
    conn.execute(
        """CREATE TABLE test_users (
        id INTEGER PRIMARY KEY,
        email TEXT NOT NULL UNIQUE,
        username TEXT NOT NULL UNIQUE,
        pwd_hash TEXT NOT NULL,
        active INTEGER,
        joined TEXT NOT NULL);
        """
    )
    yield
    conn.execute("""DROP TABLE test_users;""")


@pytest.mark.asyncio
async def test_handle_user_create(
    aiohttp_client, init_test_app, create_test_user_table
):
    payload = {
        "email": "tintin@gmail.com",
        "username": "Tintin",
        "password": "y0u != n00b1e",
    }
    client = await aiohttp_client(init_test_app)
    resp = await client.post("/users/", json=payload)
    assert resp.status == 200
    resp_json = await resp.json()
    assert resp_json["email"] == payload["email"]
    assert resp_json["username"] == payload["username"]
    assert resp_json["pwd_hash"] != payload["pwd_hash"]
    assert resp_json["active"] == 0
    client.close()

И вот полный след ошибки времени выполнения (наряду с предупреждениями об устаревании - о которых я также буду благодарен за помощь :)) 

$ pytest
============================= test session starts =============================
platform win32 -- Python 3.6.8, pytest-5.3.5, py-1.8.1, pluggy-0.13.1
rootdir: C:\Users\Mfonism\Codeville\AIOHttp\curious_me
plugins: aiohttp-0.3.0, asyncio-0.10.0
collected 1 item

test_app.py F                                                            [100%]

================================== FAILURES ===================================
_______________________ test_handle_user_create[pyloop] _______________________

aiohttp_client = <function aiohttp_client.<locals>.go at 0x0000015DE239AD08>
init_test_app = <Application 0x15de23a0d30>, create_test_user_table = None

    @pytest.mark.asyncio
    async def test_handle_user_create(
        aiohttp_client, init_test_app, create_test_user_table
    ):
        payload = {
            "email": "tintin@gmail.com",
            "username": "Tintin",
            "password": "y0u != n00b1e",
        }
        client = await aiohttp_client(init_test_app)
>       resp = await client.post("/users/", json=payload)

test_app.py:89:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
env36\lib\site-packages\aiohttp\test_utils.py:295: in request
    method, self.make_url(path), **kwargs
env36\lib\site-packages\aiohttp\client.py:417: in _request
    with timer:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

self = <aiohttp.helpers.TimerContext object at 0x0000015DE23A3B38>

    def __enter__(self) -> BaseTimerContext:
        task = current_task(loop=self._loop)

        if task is None:
>           raise RuntimeError('Timeout context manager should be used '
                               'inside a task')
E           RuntimeError: Timeout context manager should be used inside a task

env36\lib\site-packages\aiohttp\helpers.py:568: RuntimeError
============================== warnings summary ===============================
test_app.py::test_handle_user_create[pyloop]
  c:\users\mfonism\codeville\aiohttp\curious_me\env36\lib\site-packages\aiohttp\cookiejar.py:55: DeprecationWarning: The object should be created from async function
    super().__init__(loop=loop)

test_app.py::test_handle_user_create[pyloop]
  c:\users\mfonism\codeville\aiohttp\curious_me\env36\lib\site-packages\aiohttp\test_utils.py:247: DeprecationWarning: The object should be created from async function
    **kwargs)

test_app.py::test_handle_user_create[pyloop]
  c:\users\mfonism\codeville\aiohttp\curious_me\env36\lib\site-packages\aiohttp\connector.py:730: DeprecationWarning: The object should be created from async function
    loop=loop)

test_app.py::test_handle_user_create[pyloop]
  c:\users\mfonism\codeville\aiohttp\curious_me\env36\lib\site-packages\aiohttp\connector.py:735: DeprecationWarning: The object should be created from async function
    resolver = DefaultResolver(loop=self._loop)

-- Docs: https://docs.pytest.org/en/latest/warnings.html
====== 1 failed, 4 warnings in 0.78s ======
...