使用unittest.IsolatedAsyncioTestCase遇未关闭套接字ResourceWarning

问题根源分析 #

出现未关闭socket和_SelectorSocketTransport警告的核心原因有两个：

1. ClientSession上下文管理不规范：原代码中直接调用await ClientSession().__aenter__()获取会话，但未保留上下文管理器引用，仅手动调用close()无法触发完整的资源清理流程（aiohttp的close()是异步发起关闭，需要等待内部任务完成）。
2. 响应体未释放：获取ClientResponse后未读取响应体，aiohttp会保持连接直到超时，测试结束过快时就会触发资源警告。
3. 测试事件循环清理不及时：IsolatedAsyncioTestCase的测试结束后，事件循环可能被快速销毁，导致aiohttp的连接回收任务未执行完毕。

解决方案 #

方案1：修复Scraper的上下文管理器，正确托管ClientSession

修改Scraper类，完整使用ClientSession的上下文管理器生命周期，确保资源被正确清理：

import unittest
import asyncio
from aiohttp import ClientSession, ClientResponse

COUNT_OF_TESTS = 20

class Scraper:
    async def do_get_request(self, url: str) -> ClientResponse:
        resp = await self.session.get(url)
        # 必须读取响应体，释放连接到池或关闭
        await resp.read()
        return resp

    async def __aenter__(self) -> 'Scraper':
        # 保存上下文管理器引用，用于后续完整退出
        self._session_ctx = ClientSession().__aenter__()
        self.session = await self._session_ctx
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
        # 通过上下文管理器的__aexit__完成完整清理
        await self._session_ctx.__aexit__(exc_type, exc_val, exc_tb)

class TestScraper(unittest.IsolatedAsyncioTestCase):
    async def asyncTearDown(self) -> None:
        # 给事件循环留时间完成资源清理任务
        await asyncio.sleep(0.1)

    async def _some_test(self) -> None:
        async with Scraper() as scraper:
            resp = await scraper.do_get_request('https://icanhazip.com')
        self.assertEqual(resp.status, 200)

for i in range(1, COUNT_OF_TESTS+1):
    setattr(TestScraper, f'test_{i}', TestScraper._some_test)

方案2：测试类级别共享ClientSession（更高效）

如果测试逻辑允许，在测试类的asyncSetUpClass和asyncTearDownClass中创建/销毁单一ClientSession，减少资源创建频率：

import unittest
from aiohttp import ClientSession, ClientResponse

COUNT_OF_TESTS = 20

class Scraper:
    def __init__(self, session: ClientSession):
        self.session = session

    async def do_get_request(self, url: str) -> ClientResponse:
        resp = await self.session.get(url)
        await resp.read()
        return resp

class TestScraper(unittest.IsolatedAsyncioTestCase):
    @classmethod
    async def asyncSetUpClass(cls) -> None:
        cls._session_ctx = ClientSession().__aenter__()
        cls.session = await cls._session_ctx

    @classmethod
    async def asyncTearDownClass(cls) -> None:
        await cls._session_ctx.__aexit__(None, None, None)

    async def _some_test(self) -> None:
        scraper = Scraper(self.session)
        resp = await scraper.do_get_request('https://icanhazip.com')
        self.assertEqual(resp.status, 200)

for i in range(1, COUNT_OF_TESTS+1):
    setattr(TestScraper, f'test_{i}', TestScraper._some_test)

关键注意点 #

• 必须读取响应体：无论是否需要响应内容，调用resp.read()/resp.text()等方法是释放连接的必要步骤，否则aiohttp会保持连接占用。
• 完整使用上下文管理器：不要手动拆分__aenter__和__aexit__的调用，通过上下文管理器的完整流程才能确保ClientSession的所有资源被清理。
• 给清理留时间：在asyncTearDown中加入短暂延迟，让事件循环有机会完成aiohttp的后台清理任务。

内容的提问来源于stack exchange，提问作者555Russich