unittest测试async函数需将测试方法写为普通函数，内部用asyncio.run()调用协程；mock异步函数必须用AsyncMock；HTTP等外部依赖应使用httpx.MockTransport等专用mock工具；注意清理事件循环任务和资源以防卡死。
async def 函数怎么用 unittest 测？直接 await 会报错
Python 的 unittest 默认不支持异步测试方法，如果在 TestCase 里写 async def test_something(self)，运行时会跳过或报 RuntimeWarning: coroutine 'test_something' was never awaited。根本原因是 unittest 的执行器不识别协程对象。
正确做法是：把测试方法保持为普通函数，内部用 asyncio.run() 或 self.loop.run_until_complete()（需手动管理事件循环）来驱动被测协程。推荐前者，简洁且避免循环复用问题：
import asyncio
import unittest
<p>class TestMyAsyncFunc(unittest.TestCase):
def test_fetch_data(self):
result = asyncio.run(fetch_data("<a href="https://www.php.cn/link/4d2fe2e8601f7a8018594d98f28706f2">https://www.php.cn/link/4d2fe2e8601f7a8018594d98f28706f2</a>"))
self.assertIn("json", result)</p>注意：asyncio.run() 每次都新建事件循环，适合隔离性要求高的单元测试；但不能在已有运行中的循环里调用（比如 Jupyter 或某些测试框架嵌套场景），此时得用 asyncio.get_event_loop().run_until_complete() 并确保循环未关闭。
如何 mock 异步函数？AsyncMock 是唯一可靠选择
用普通 Mock 替换 async def 函数会导致调用时报 TypeError: object Mock can't be used in 'await' expression。必须用 AsyncMock —— 它是 Python 3.8+ unittest.mock 内置的专用类型，能被 await 且支持 return_value、side_effect 等常规行为。
立即学习“Python免费学习笔记（深入）”；
-
AsyncMock返回的是协程对象，不是直接值；若设mock.return_value = "ok"，实际需要await mock()才得到"ok" - 若要让 mock 抛出异常，用
mock.side_effect = ValueError("boom")，它会在await mock()时触发 - 对类方法打 patch，记得 target 路径写对，比如
@patch("myapp.client.AsyncClient.get")，而不是AsyncClient.get的实例方法路径
示例：
from unittest.mock import AsyncMock, patch
<p>@patch("myapp.api.fetch_user")
def test_get_user_profile(self, mock_fetch):
mock_fetch.return_value = {"id": 123, "name": "alice"}</p><h1>被测函数是 async def get_profile(...)，内部 await fetch_user()</h1><pre class='brush:python;toolbar:false;'>result = asyncio.run(get_profile(123))
self.assertEqual(result["name"], "alice")测试依赖外部 HTTP 或数据库怎么办？别真连，用 httpx.MockTransport 或 aiomock
真实网络调用会让测试变慢、不稳定、依赖环境。异步 I/O 类库（如 httpx、aiohttp、asyncpg）都有配套的 mock 方式，比通用 AsyncMock 更贴近真实行为。
以 httpx 为例，优先用 httpx.MockTransport：
import httpx
<p>transport = httpx.MockTransport(
lambda request: httpx.Response(200, json={"data": "test"})
)
client = httpx.AsyncClient(transport=transport)
result = asyncio.run(client.get("<a href="https://www.php.cn/link/a1d742d648b1409658be0c7b97e9e743">https://www.php.cn/link/a1d742d648b1409658be0c7b97e9e743</a>"))</p>关键点：
-
MockTransport拦截的是整个请求生命周期，包括连接、重试、超时逻辑，比只 mockget方法更彻底 - 不要用
patch去 mockhttpx.AsyncClient.get—— 容易漏掉底层连接池、SSL 验证等副作用 - 对
aiohttp，可用aioresponses库；对asyncpg，可用aiomock或内存级postgresql://localhost/testdb+ 清库 fixture
为什么测 async 函数总卡住或超时？检查事件循环和资源清理
常见卡死现象：测试挂起几秒后报 asyncio.TimeoutError 或直接无响应。大概率是以下其一：
- 被测代码启动了后台任务（如
asyncio.create_task(...)）但没等待完成，测试主线程结束时任务还在跑 - 用了
asyncio.sleep(3)这类硬等待，应替换为可注入的延迟参数或用pytest-asyncio的event_loopfixture 控制时间 - mock 对象没正确 await，比如写了
mock_fn()却忘了await，导致返回协程对象未执行 - 数据库连接池未 close，下个测试用到同一 loop 时因连接耗尽而阻塞
建议在每个测试末尾加显式清理：
def tearDown(self):
# 清空所有 pending tasks
pending = asyncio.all_tasks()
if pending:
loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.gather(*pending, return_exceptions=True))
真正难调试的，往往是跨协程状态残留 —— 比如一个全局缓存字典被多个测试修改后没重置。这种必须靠 fixture 级隔离，不能只靠 mock。
