pytest+asyncio+SQLAlchemy异步测试异常：首测通过次测失败

我在测试一个从PostgreSQL数据库根据邮箱地址获取User对象的异步函数get_user（定义于user.py）。单个测试用例test_get_user_one能正常通过，但新增完全相同的test_get_user_two时，第二个测试会抛出以下错误：

...
...
>               self._write_fut = self._loop._proactor.send(self._sock, data)
E               AttributeError: 'NoneType' object has no attribute 'send'

C:\Python\lib\asyncio\proactor_events.py:397: AttributeError
================================ short test summary info ===========================================
FAILED test/test_user.py::test_get_user_two - AttributeError: 'NoneType' object has no attribute 'send'
================================ 1 failed, 1 passed in 1.02s =======================================

交换两个测试的执行顺序后，先执行的测试通过，后执行的测试依然报相同错误。

相关代码 #

session.py #

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

async_engine = create_async_engine("postgresql+asyncpg://url", echo=False,future=True)
async_session = sessionmaker(async_engine,class_=AsyncSession,expire_on_commit=False)

user.py #

from sqlalchemy.ext.asyncio import AsyncSession
from model import User

async def get_user(email: str, session: AsyncSession) -> User:
    query = select(User).filter(User.email == email)
    async with session:
        results = await session.execute(query)
        user = results.scalar_one_or_none()
        return user

test_user.py #

from session import async_session
from user import get_user

@pytest.mark.asyncio
async def test_get_user_one():
    user = await get_user(_email, async_session())
    assert user.email == "user@email.com"

@pytest.mark.asyncio
async def test_get_user_two():
    user = await get_user(_email, async_session())
    assert user.email == "user@email.com"

model.py #

from typing import List, Optional
from pydantic import EmailStr
from sqlmodel import Column, Field, Relationship, SQLModel, String

class UserBase(SQLModel):
    email: EmailStr = Field(sa_column=Column("email", String, unique=True))

class User(UserBase, table=True):
    id: Optional[int] = Field(default=None, primary_key=True)
    lastname: str | None
    firstname: str | None

问题排查与修复方案 #

核心原因 #

get_user函数里使用了async with session:，这个语法会在代码块执行完毕后自动关闭传入的session。而你在测试中直接把async_session()创建的session传给函数，第一个测试结束后session被关闭，连接资源被销毁，第二个测试再尝试使用相关连接时，就会出现sock为None的错误。

修复步骤 #

1. 修改get_user函数，移除内部的session管理：
   把函数里的async with session:去掉，由调用方负责session的创建和关闭，避免函数意外关闭外部传入的session：

   from sqlalchemy.ext.asyncio import AsyncSession
   from sqlmodel import select  # 补充select的导入，原代码遗漏了
   from model import User
   
   async def get_user(email: str, session: AsyncSession) -> User:
       query = select(User).filter(User.email == email)
       results = await session.execute(query)
       user = results.scalar_one_or_none()
       return user
   

2. 在测试用例中自行管理session生命周期：
   每个测试用例使用async with创建并关闭session，确保每个测试用的session都是独立且正确释放的：

   from session import async_session
   from user import get_user
   
   @pytest.mark.asyncio
   async def test_get_user_one():
       async with async_session() as session:
           user = await get_user("user@email.com", session)
           assert user.email == "user@email.com"
   
   @pytest.mark.asyncio
   async def test_get_user_two():
       async with async_session() as session:
           user = await get_user("user@email.com", session)
           assert user.email == "user@email.com"
   

3. 额外优化：使用pytest fixture复用session逻辑：
   可以定义一个fixture来统一创建session，减少测试代码重复：

   import pytest
   from session import async_session
   
   @pytest.fixture
   async def db_session():
       async with async_session() as session:
           yield session
   
   @pytest.mark.asyncio
   async def test_get_user_one(db_session):
       user = await get_user("user@email.com", db_session)
       assert user.email == "user@email.com"
   
   @pytest.mark.asyncio
   async def test_get_user_two(db_session):
       user = await get_user("user@email.com", db_session)
       assert user.email == "user@email.com"
   

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