Что я делаю
Я изучаю 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 ======