如何检查是否从未等待过任何异步函数?

huangapple go评论65阅读模式
英文:

How to check if any async function has never been awaited?

问题

以下是您要翻译的代码部分:

def checkawait(test):
    @functools.wraps(test)
    async def wrapper(*args, **kwargs):
        coros = []
        res = test(*args, **kwargs)
        if asyncio.iscoroutine(res):
            coros.append(res)
        while coros:
            done, coros = await asyncio.wait(coros, timeout=0.1)
            if not done:
                raise Exception("Not all coroutines have completed")
        return res
    return wrapper

希望这对您有所帮助。如果您需要进一步的解释或帮助,请随时告诉我。

英文:

I'm testing my Python software using Pytest.
I have to call many async function and sometimes my tests pass even when I forgot to write the await keyword.

I would like my test to automatically fail if I call an async function without await.

I was thinking about a decorator to add at the top of my tests, something like

async def func():
    return 42

@checkawait
async def test_await():
    func() # <-- forgot to await

I would like this test to fail with the decorator, because func is an async function that was never awaited.
(I know this is not a proper test, since I'm not testing anything. It's just an example).

Without the decorator test_await passes.

I really don't know what to do.

I asked chatGPT which told me to use this

def checkawait(test):
    @functools.wraps(test)
    async def wrapper(*args, **kwargs):
        coros = []
        res = test(*args, **kwargs)
        if asyncio.iscoroutine(res):
            coros.append(res)
        while coros:
            done, coros = await asyncio.wait(coros, timeout=0.1)
            if not done:
                raise Exception("Not all coroutines have completed")
        return res
    return wrapper

which, of course, is not working.

Does anyone have any ideas? Is it even possible to do so?
Thanks

答案1

得分: 0

不等待协程会生成RuntimeWarning警告。
运行pytest时,使用pytest test.py -W error::RuntimeWarning,我们会得到一个PytestUnraisableExceptionWarning警告,但测试仍然通过。

我考虑了以下解决方案

import functools
import warnings

def checkawait(my_test):
    @functools.wraps(my_test)
    async def wrapper(*args, **kwargs):
        with pytest.warns(RuntimeWarning) as record:
            await my_test(*args, **kwargs)
            captured_warnings = record.list
            if not captured_warnings:
                warnings.warn("无害警告", RuntimeWarning)
        for warning in captured_warnings:
            if "never awaited" in warning.message.args[0]:
                raise Exception(warning.message.args[0])
    return wrapper

作为装饰器使用,checkawait失败第一个测试但通过第二个测试:

async def func():
    return 42

@checkawait
async def test_fail():
    func()

@checkawait
async def test_pass():
    await func()
英文:

Not awaited coroutines generate RuntimeWarnings.
Running pytest as pytest test.py -W error::RuntimeWarning, we get a PytestUnraisableExceptionWarning and the test still pass.

I thought of this solution

import functools
import warnings

def checkawait(my_test):
    @functools.wraps(my_test)
    async def wrapper(*args, **kwargs):
        with pytest.warns(RuntimeWarning) as record:
            await my_test(*args, **kwargs)
            captured_warnings = record.list
            if not captured_warnings:
                warnings.warn("Harmless warning", RuntimeWarning)
        for warning in captured_warnings:
            if "never awaited" in warning.message.args[0]:
                raise Exception(warning.message.args[0])
    return wrapper

Used as a decorator, checkawait fails the first test but passes the second one:

async def func():
    return 42

@checkawait
async def test_fail():
    func()

@checkawait
async def test_pass():
    await func()

huangapple
  • 本文由 发表于 2023年2月24日 00:18:21
  • 转载请务必保留本文链接:https://go.coder-hub.com/75547485.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定