Python高级技术之：`pytest-asyncio`：如何测试异步`Python`代码。 - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

各位观众老爷，大家好！我是你们的老朋友，今天咱来聊聊Python异步代码的测试，特别是用pytest-asyncio这玩意儿。保证让各位听完之后，腰不酸了，腿不疼了，测试异步代码也更有劲儿了！

Part 1: 异步编程的那些事儿

首先，咱得稍微回顾一下异步编程。为啥要有异步？简单来说，就是为了让你的程序在等待某些耗时操作（比如网络请求、数据库查询）的时候，别傻乎乎地干等着。它可以去干点别的，等数据回来了再回来处理。这样就能提高程序的效率。

Python里实现异步编程，主要靠asyncio库。它引入了async和await这两个关键字。async用来声明一个协程函数，await用来等待一个协程函数完成。

举个例子，假设我们要异步地从两个网站获取数据：

import asyncio
import aiohttp

async def fetch_url(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

async def main():
    url1 = "https://www.example.com"
    url2 = "https://www.python.org"

    task1 = asyncio.create_task(fetch_url(url1))
    task2 = asyncio.create_task(fetch_url(url2))

    result1 = await task1
    result2 = await task2

    print(f"Result from {url1}: {result1[:100]}...") # 截取前100个字符
    print(f"Result from {url2}: {result2[:100]}...")

if __name__ == "__main__":
    asyncio.run(main())

在这个例子里，fetch_url函数是一个协程函数，它使用aiohttp库异步地获取网页内容。main函数也声明为协程函数，它创建了两个任务task1和task2，分别获取两个网站的内容，然后使用await等待它们完成。

Part 2: 异步代码的测试难题

现在问题来了，异步代码怎么测试？传统的测试方法可能不太灵光了。因为异步函数通常需要在事件循环中运行，而且我们可能需要模拟一些异步操作，比如网络请求的延迟。

如果我们直接调用一个异步函数，它不会立即执行，而是返回一个协程对象。我们需要把它提交到事件循环中才能真正运行。

import asyncio

async def my_async_function():
    await asyncio.sleep(1)
    return "Hello, async world!"

# 错误的做法：直接调用异步函数
# result = my_async_function()
# print(result)  # 输出：<coroutine object my_async_function at 0x...>

# 正确的做法：在事件循环中运行
async def main():
    result = await my_async_function()
    print(result)

if __name__ == "__main__":
    asyncio.run(main())

Part 3: pytest-asyncio闪亮登场

pytest-asyncio就是来解决这个问题的。它是一个pytest插件，专门用来测试异步Python代码。它可以自动地为你的测试函数创建事件循环，并且提供了一些有用的工具函数，让你更容易地模拟异步操作。

安装pytest-asyncio

pip install pytest pytest-asyncio

基本用法

首先，我们需要在pytest.ini或者pyproject.toml文件中配置pytest-asyncio：

# pytest.ini
[pytest]
asyncio_mode = auto

asyncio_mode = auto 表示让pytest-asyncio自动检测哪些测试函数是异步的，然后自动为它们创建事件循环。你也可以选择其他模式，比如strict，它会强制你显式地标记异步测试函数。

接下来，我们可以编写一个异步测试函数：

import pytest
import asyncio

async def my_async_function():
    await asyncio.sleep(0.1)
    return "Hello, async world!"

@pytest.mark.asyncio
async def test_my_async_function():
    result = await my_async_function()
    assert result == "Hello, async world!"

注意，我们用@pytest.mark.asyncio装饰器标记了test_my_async_function函数。这样pytest-asyncio就知道这是一个异步测试函数，会为它创建一个事件循环。

更高级的用法

pytest-asyncio还提供了一些更高级的功能，比如：

异步fixture: 可以创建异步的fixture，用于在测试函数中共享资源。
模拟异步操作: 可以使用asyncio.sleep等函数模拟异步操作的延迟。
自定义事件循环: 可以创建自定义的事件循环，用于更精细的控制。

Part 4: 异步Fixture的妙用

Fixture 是 pytest 中一个非常强大的特性，它可以帮助我们在测试函数执行之前准备好测试环境，或者在测试函数执行之后清理测试环境。pytest-asyncio 允许我们创建异步的 fixture，这对于测试异步代码来说非常有用。

例如，我们可以创建一个异步的 fixture 来初始化一个异步的数据库连接：

import pytest
import asyncio
import aiosqlite

@pytest.fixture
async def async_db():
    async with aiosqlite.connect(":memory:") as db:
        await db.execute("CREATE TABLE users (id INTEGER PRIMARY KEY, name TEXT)")
        await db.execute("INSERT INTO users (name) VALUES (?)", ("Alice",))
        await db.commit()
        yield db
        await db.close()

@pytest.mark.asyncio
async def test_async_db(async_db):
    async with async_db.execute("SELECT * FROM users") as cursor:
        rows = await cursor.fetchall()
        assert len(rows) == 1
        assert rows[0][1] == "Alice"

在这个例子中，async_db 是一个异步的 fixture。它使用 aiosqlite 库连接到一个内存数据库，创建一个 users 表，并插入一条数据。yield db 语句会将数据库连接对象传递给测试函数。在测试函数执行完毕后，async with 语句会自动关闭数据库连接。

Part 5: 模拟异步操作

在测试异步代码时，我们经常需要模拟一些异步操作的延迟或者失败。pytest-asyncio 并没有提供专门的模拟工具，但是我们可以使用 asyncio.sleep 函数来模拟延迟，或者使用 pytest.raises 上下文管理器来模拟异常。

例如，我们可以模拟一个网络请求的延迟：

import pytest
import asyncio

async def fetch_data(url):
    await asyncio.sleep(0.5)  # 模拟网络延迟
    return f"Data from {url}"

@pytest.mark.asyncio
async def test_fetch_data():
    result = await fetch_data("https://www.example.com")
    assert result == "Data from https://www.example.com"

在这个例子中，fetch_data 函数使用 asyncio.sleep(0.5) 模拟了 0.5 秒的网络延迟。

我们也可以模拟一个网络请求失败的情况：

import pytest
import asyncio

async def fetch_data(url):
    await asyncio.sleep(0.1)
    raise aiohttp.ClientError("Network error")

@pytest.mark.asyncio
async def test_fetch_data_error():
    with pytest.raises(aiohttp.ClientError) as excinfo:
        await fetch_data("https://www.example.com")
    assert "Network error" in str(excinfo.value)

在这个例子中，fetch_data 函数会抛出一个 aiohttp.ClientError 异常。我们使用 pytest.raises 上下文管理器来捕获这个异常，并验证异常信息是否正确。

Part 6: 自定义事件循环

在某些情况下，我们可能需要创建自定义的事件循环，用于更精细的控制。pytest-asyncio 允许我们通过 fixture 来提供自定义的事件循环。

例如，我们可以创建一个 fixture 来提供一个 uvloop 事件循环：

import pytest
import asyncio
import uvloop

@pytest.fixture(scope="session")
def event_loop():
    loop = uvloop.new_event_loop()
    asyncio.set_event_loop(loop)
    yield loop
    loop.close()

@pytest.mark.asyncio
async def test_uvloop():
    print(f"Running test in event loop: {asyncio.get_event_loop()}")
    assert isinstance(asyncio.get_event_loop(), uvloop.Loop)

在这个例子中，event_loop fixture 创建了一个 uvloop 事件循环，并将其设置为当前事件循环。scope="session" 表示这个 fixture 在整个测试会话中只会被创建一次。

Part 7: 常见问题与注意事项

忘记标记异步测试函数: 如果你忘记用 @pytest.mark.asyncio 标记异步测试函数，pytest 会把它当成一个普通的同步函数来运行，导致测试失败。
阻塞事件循环: 避免在异步测试函数中执行阻塞操作，比如 time.sleep。这会导致事件循环被阻塞，影响测试的性能。应该使用 asyncio.sleep 来模拟延迟。
资源泄漏: 确保在测试函数执行完毕后，正确地释放所有资源，比如关闭数据库连接、关闭文件句柄等。可以使用 async with 语句或者 fixture 来自动管理资源。
小心死锁: 在编写异步代码时，要小心死锁的发生。死锁是指两个或多个协程互相等待对方释放资源，导致程序无法继续执行。可以使用 asyncio.wait_for 函数来设置超时时间，避免无限等待。

Part 8: 实战案例：测试一个异步HTTP服务

假设我们有一个简单的异步 HTTP 服务，它使用 aiohttp 库来处理请求：

from aiohttp import web
import asyncio

async def handle(request):
    await asyncio.sleep(0.1)  # 模拟处理请求的延迟
    name = request.match_info.get('name', "Anonymous")
    text = f"Hello, {name}"
    return web.Response(text=text)

async def create_app():
    app = web.Application()
    app.add_routes([web.get('/', handle),
                     web.get('/{name}', handle)])
    return app

async def main():
    app = await create_app()
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()
    print("Server started on http://localhost:8080")
    # Keep the server running indefinitely, usually replaced with signal handling
    # await asyncio.Future()
    return runner #返回runner是为了方便测试关闭

if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("Server stopped.")

我们可以使用 pytest-asyncio 来测试这个服务。首先，我们需要创建一个 fixture 来启动和停止服务：

import pytest
import asyncio
from aiohttp import web
import aiohttp
from your_module import create_app, main  # 替换为你的模块名

@pytest.fixture
async def server(event_loop): #依赖内置的event_loop fixture，保证同步
    app = await create_app()
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, 'localhost', 8080)
    await site.start()

    yield  # Provide the running server to the tests

    await runner.cleanup() # 关闭服务

@pytest.mark.asyncio
async def test_hello_world(server):
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:8080") as resp:
            assert resp.status == 200
            text = await resp.text()
            assert "Hello, Anonymous" in text

@pytest.mark.asyncio
async def test_hello_name(server):
    async with aiohttp.ClientSession() as session:
        async with session.get("http://localhost:8080/John") as resp:
            assert resp.status == 200
            text = await resp.text()
            assert "Hello, John" in text

在这个例子中，server fixture 启动了一个 aiohttp 服务，并将其运行在 localhost:8080 上。测试函数使用 aiohttp.ClientSession 来发送 HTTP 请求，并验证响应的状态码和内容。

Part 9: 总结

pytest-asyncio 是一个非常强大的工具，它可以让你更容易地测试异步 Python 代码。通过使用异步 fixture、模拟异步操作和自定义事件循环，你可以编写出更可靠、更高效的异步测试。记住，测试是保证代码质量的关键，不要偷懒哦！

好了，今天的讲座就到这里。希望各位有所收获，以后再见！

发表回复 取消回复

发表回复取消回复