5

我在做什么

我正在通过构建一个 REST api 来学习 aiohttp，我正在使用 Pytest（及其 async 和 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

从错误消息中，我了解到客户端正在尝试使用异步超时上下文管理器，但这失败了，因为我没有在任务中调用它。

我不知道我的推论是否正确。

而且，我对 asyncio 不太满意，不知道如何解决这个问题。

如果有人告诉我出路，我将不胜感激。

更多信息

这是我的测试文件的源代码：

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

0 回答 0