标签: pytest-asyncio

@pytest.fixture
def d_service():
    c = DService()
    return c

# @pytest.mark.asyncio  # tried it too
async def test_get_file_list(d_service):
    files = await d_service.get_file_list('')
    print(files)

然而，却出现了以下错误？

已收集 0 项 / 1 错误

===================================== 错误============== =====================
________________ 收集测试/e2e_tests/test_d.py 时出错 _________________
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:617: 在 __call__ 中
    返回 self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:222: 在 _hookexec 中
    返回 self._inner_hookexec(钩子、方法、kwargs)
..\..\..\..\..\anaconda3\lib\site-packages\pluggy\__init__.py:216：在
    第一个结果=hook.spec_opts.get('第一个结果'),
..\..\..\..\..\anaconda3\lib\site-packages\_pytest\python.py:171：在 pytest_pycollect_makeitem 中
    res = 结果.get_result()
..\..\..\..\..\anaconda3\lib\site-packages\anyio\pytest_plugin.py:98：在 pytest_pycollect_makeitem 中
    标记=collector.get_closest_marker('anyio')
E AttributeError：“模块”对象没有属性“get_closest_marker”
!!!!!!!!!!!!!!!!!!!!! 已中断：收集期间出现 1 个错误!!!!!!!!!!!!!!!!!!
=========================== 2.53 秒内出现 1 个错误 ================== =========

我安装了以下软件包。错误消失，但测试被跳过。

pip install …

我有一个测试来验证异步响应是否抛出异常，我正在使用 pytest-asyncio 版本 0.10.0 来运行该测试。

代码基本上是：

class TestThis:
    @pytest.mark.asyncio
    def test_the_thing(self):
       arg1 = "cmd"
       arg2 = "second command"
       with pytest.raises(CustomException):
           await do_thing(arg1, arg2)

现在真正奇怪的是，如果我单独运行这个测试，或者单独运行这个类，这个测试就可以正常工作。但是，当我运行所有测试（项目根目录下的 pytest）时，每次都会失败并出现运行时错误，表示循环已关闭。

我正在尝试为我的 fastapi 应用程序编写一些测试

\n

我正在使用prisma-client-py数据库。我不知道这是否会改变什么

\n

一切都按预期工作，除了第一个和最后一个之外，它们都因相同的错误而失败：

RuntimeError: <asyncio.locks.Event object at 0x7f5696832950 [unset]> is bound to a different event loop\n

这是我的conftest.py

import os\nimport asyncio\nimport pytest\nfrom typing import Any, AsyncGenerator, Generator, Iterator\n\nfrom fastapi import FastAPI\nfrom fastapi.testclient import TestClient\nfrom prisma import Prisma, register\n\n\nfrom server.database.base import *\nfrom server.config.exceptions import configure_exception_handlers\nfrom server.config.settings import settings\nfrom server.apis import apis\n\n\ndef start_application() -> FastAPI:\n    """\n    Return a FastAPI app\n    """\n    _app = FastAPI(\n        title=str(settings.TITLE),\n        description=str(settings.DESCRIPTION),\n        version=str(settings.VERSION),\n    )\n    configure_exception_handlers(_app)\n    _app.include_router(apis)\n    return …

我正在尝试构建FastAPI完全覆盖测试的应用程序python 3.9 为此目的，我选择了堆栈：FastAPI、uvicorn、SQLAlchemy、asyncpg、pytest（+ async、cov 插件）、覆盖范围和 httpx AsyncClient

这是我的最低要求.txt

所有测试都运行顺利，我得到了预期的结果。但我遇到了问题，覆盖范围没有正确收集。await当协程将控制权返回到事件循环时，它会在第一个关键字之后中断

以下是关于如何重现此行为的最小集合（也可以在 GitHub 上找到）。

应用代码main.py：

import sqlalchemy as sa
from fastapi import FastAPI
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from starlette.requests import Request

app = FastAPI()
DATABASE_URL = 'sqlite+aiosqlite://?cache=shared'


@app.on_event('startup')
async def startup_event():
    engine = create_async_engine(DATABASE_URL, future=True)
    app.state.session = AsyncSession(engine, expire_on_commit=False)
    app.state.engine = engine


@app.on_event('shutdown')
async def shutdown_event():
    await app.state.session.close()


@app.get('/', name="home")
async def get_home(request: Request):
    res = await request.app.state.session.execute(sa.text('SELECT 1'))
    # after …

我正在测试可能会陷入僵局的异步功能。我试图添加一个固定装置来限制该功能在引发故障之前只能运行5秒钟，但到目前为止还没有奏效。

设定：

pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0

码：

pipenv --python==3.6
pipenv install pytest==4.4.1
pipenv install pytest-asyncio==0.10.0

-

编辑：Mikhail的解决方案工作正常。但是，我找不到将其合并到灯具中的方法。

测试以下代码时

@pytest.mark.asynico
async def test_handle_DATA(mocker):
    handle_mock = mocker.MagicMock()
    envelope_mock = mocker.MagicMock(mail_from="Test@From", rcpt_tos=["Test@To"], content=b"TestContent")

    result = SendToDictHandler.handle_DATA(handle_mock, "TestServer", "TestSession", envelope_mock)

    assert result == "250 Message accepted for delivery"
    assert email_core.testing_emails_dict == {
        "Test@To": {
            "from": "Test@From",
            "to": ["Test@To"],
            "msg": "TestContent",
        }
    }

pytest -vvv在项目环境中运行时收到的警告：

PytestWarning: Coroutine functions are not natively supported and have been skipped.
You need to install a suitable plugin for your async framework, for example:
 - pytest-asyncio
 - pytest-trio
 - pytest-tornasync
warnings.warn(PytestWarning(msg.format(pyfuncitem.nodeid)))

我确实pytest-asyncio安装了。我通过pytest --trace-config …

我想使用 asyncio (/curio/trio) 和 pytest 同时运行多个测试，但我找不到任何相关信息。我需要自己安排吗？如果我这样做，有没有办法有一个很好的输出来分离（子）测试用例？

这是我正在尝试的一个小玩具示例：

import pytest
import time
import asyncio

pytestmark = pytest.mark.asyncio
expected_duration = 1
accepted_error = 0.1

async def test_sleep():
    start = time.time()
    time.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

async def test_async_sleep():
    start = time.time()
    await asyncio.sleep(expected_duration)
    duration = time.time() - start
    assert abs(duration-expected_duration) < accepted_error

我已经尝试了所有我能找到的东西，但我无法让异步测试工作。

我开始RuntimeError: This event loop is already running运行TestClient（根据文档，这是有意义的），但我开始httpx.ConnectError: [Errno 8] nodename nor servname provided, or not known使用httpx AsyncClient.

我有一个简单的测试：

@pytest.fixture(scope="module")
async def async_client() -> Generator:
    async with AsyncClient(app=app, base_url='http://0.0.0.0') as client:
        yield client

@pytest.mark.asyncio@mock.patch('apps.core.views.requests.get', new=mocked_get_request)
async def test_get_non_json_response(async_client: AsyncClient):
    response = await async_client.get("/mymedia")
    assertEqual(response.json()['error']['message']['message'], 'Not json')

哪里 /media：

@app.get('/mymedia')
async def my_media(request: Request, cache: RedisCacheBackend = Depends(redis_cache)):
    return await my_media_ep(request, cache=cache)

my_media_ep是一个包含多个异步 api 调用的长函数。

我也按照异步测试文档中的建议进行了尝试，但得到了相同的错误。

有什么建议或例子吗？

如何定义异步装置并在异步测试中使用它们？

以下代码全部在同一个文件中,失败了.是否由测试跑者明确地称之为夹具并且没有等待？

@pytest.fixture
async def create_x(api_client):
    x_id = await add_x(api_client)
    return api_client, x_id

async def test_app(create_x, auth):
    api_client, x_id = create_x
    resp = await api_client.get(f'my_res/{x_id}', headers=auth)
    assert resp.status == web.HTTPOk.status_code

生产

==================================== ERRORS ====================================
_____________ ERROR at setup of test_app[pyloop] ______________

api_client = <aiohttp.test_utils.TestClient object at 0x7f27ec954f60>

    @pytest.fixture
    async def create_x(api_client):
>       x_id = await add_x(api_client)
...
... cannot show the full trace and pathnames sorry
...    

in __await__
    ret = yield from self._coro /home/mbb/.pyenv/versions/3.6.3/envs/mr/lib/python3.6/site-packages/aiohttp/test_utils.py:245: in request
    method, self.make_url(path), *args, …

我收到错误

运行时错误：事件循环已关闭

每次我尝试在测试中进行多个异步调用时。我已经尝试使用其他 Stack Overflow 帖子中的所有其他建议来重写固定event_loop装置，但没有任何效果。我想知道我错过了什么？

运行测试命令：

python -m pytest tests/ --asyncio-mode=auto

要求.txt

pytest==7.1.2
pytest-asyncio==0.18.3
pytest-html==3.1.1
pytest-metadata==2.0.1

测试.py

async def test_user(test_client_fast_api):
    assert 200 == 200

    # works fine
    request_first = test_client_fast_api.post("/first_route")

    # recieve RuntimeError: Event loop is closed
    request_second = test_client_fast_api.post("/second_route")

测试.py

@pytest.fixture()
def event_loop():
    try:
        loop = asyncio.get_running_loop()
    except RuntimeError:
        loop = asyncio.new_event_loop()
    yield loop
    loop.close()