Pytest+FastAPI+SQLAlchemy+Postgres InterfaceError
我在使用FastAPI+SQLAlchemy和PostgreSQL运行测试时遇到了一些问题，这会导致很多错误(然而，它在SQLite上运行得很好)。我已经用MVP应用程序和Pytest在Docker组合测试上创建了一个回购程序。
基本错误是sqlalchemy.exc.InterfaceError('cannot perform operation: another operation is in progress')。这可能与app/DB初始化有关，尽管我检查了所有操作都是按顺序执行的。此外，我还尝试使用TestClient的单个实例进行所有测试，但没有得到更好的结果。我希望找到一个解决方案，一个正确的方法来测试这样的应用程序。
下面是代码中最重要的部分：
app.py：
app = FastAPI()
some_items = dict()
@app.on_event("startup")
async def startup():
await create_database()
# Extract some data from env, local files, or S3
some_items["pi"] = 3.1415926535
some_items["eu"] = 2.7182818284
@app.post("/{name}")
async def create_obj(name: str, request: Request):
data = await request.json()
if data.get("code") in some_items:
data["value"] = some_items[data["code"]]
async with async_session() as session:
async with session.begin():
await create_object(session, name, data)
return JSONResponse(status_code=200, content=data)
else:
return JSONResponse(status_code=404, content={})
@app.get("/{name}")
async def get_connected_register(name: str):
async with async_session() as session:
async with session.begin():
objects = await get_objects(session, name)
result = []
for obj in objects:
result.append({
"id": obj.id, "name": obj.name, **obj.data,
})
return resulttests.py：
@pytest.fixture(scope="module")
def event_loop():
loop = asyncio.get_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="module")
@pytest.mark.asyncio
async def get_db():
await delete_database()
await create_database()
@pytest.mark.parametrize("test_case", test_cases_post)
def test_post(get_db, test_case):
with TestClient(app)() as client:
response = client.post(f"/{test_case['name']}", json=test_case["data"])
assert response.status_code == test_case["res"]
@pytest.mark.parametrize("test_case", test_cases_get)
def test_get(get_db, test_case):
with TestClient(app)() as client:
response = client.get(f"/{test_case['name']}")
assert len(response.json()) == test_case["count"]db.py：
DATABASE_URL = environ.get("DATABASE_URL", "sqlite+aiosqlite:///./test.db")
engine = create_async_engine(DATABASE_URL, future=True, echo=True)
async_session = sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)
Base = declarative_base()
async def delete_database():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
async def create_database():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
class Model(Base):
__tablename__ = "smth"
id = Column(Integer, primary_key=True)
name = Column(String, nullable=False)
data = Column(JSON, nullable=False)
idx_main = Index("name", "id")
async def create_object(db: Session, name: str, data: dict):
connection = Model(name=name, data=data)
db.add(connection)
await db.flush()
async def get_objects(db: Session, name: str):
raw_q = select(Model) \
.where(Model.name == name) \
.order_by(Model.id)
q = await db.execute(raw_q)
return q.scalars().all()回答 1
Stack Overflow用户
发布于 2022-04-08 17:45:15
目前，测试代码非常耦合，因此测试套件的工作方式如下：
- 数据库只为所有测试创建一次。
- 第一组测试运行并填充数据库。
- 第二组测试将运行(只有在数据库完全填充时才能成功)
作为端到端的测试，这是有价值的，但我认为如果将整个程序放在一个测试函数中，它会更好。
就单元测试而言，它有一点问题。我不确定pytest-asyncio是否保证测试运行顺序(存在pytest插件只为了使测试以确定的顺序运行)，当然原则是单元测试应该相互独立。
测试也以另一种重要方式耦合--数据库I/O代码和应用程序逻辑同时进行测试。
FastAPI鼓励的做法是在路由中使用依赖注入：
from fastapi import Depends, FastAPI, Request
...
def get_sessionmaker() -> Callable:
# this is a bit baroque, but x = Depends(y) assigns x = y()
# so that's why it's here
return async_session
@app.post("/{name}")
async def create_obj(name: str, request: Request, get_session = Depends(get_sessionmaker)):
data = await request.json()
if data.get("code") in some_items:
data["value"] = some_items[data["code"]]
async with get_session() as session:
async with session.begin():
await create_object(session, name, data)
return JSONResponse(status_code=200, content=data)
else:
return JSONResponse(status_code=404, content={})当涉及到测试时，FastAPI然后是允许您交换真正的依赖项。，这样您就可以模拟数据库，并与数据库I/O隔离测试应用程序逻辑：
from app import app, get_sessionmaker
from mocks import mock_sessionmaker
...
client = TestClient(app)
...
async def override_sessionmaker():
return mock_sessionmaker
app.dependency_overrides[get_sessionmaker] = override_sessionmaker
# now we can run some tests这将意味着，当您运行测试时，无论您在mocks.mock_sessionmaker中放入什么，都会在测试中给出get_session函数，而不是get_sessionmaker。我们可以让mock_sessionmaker返回一个名为get_mock_session的函数。
换句话说，在测试中，我们使用的是with get_mock_session() as session:，而不是with get_mock_session() as session:。
不幸的是，这个get_mock_session必须返回一些有点复杂的东西(让我们称之为mock_session)，因为应用程序代码会执行async with session.begin()。
为了简单起见，我很想重构应用程序代码，但如果不是，那么在本例中调用.begin、.add和.flush时，它就不会抛出错误，而且这些方法必须是异步的。但他们不需要做任何事，所以也不算太糟.
FastAPI文档中的数据库+依赖关系的另一个示例确实使代码有点耦合，但严格地将SQLite用于单元测试，使您可以为端到端测试和应用程序本身做一些不同的事情。
https://stackoverflow.com/questions/71763595
复制相似问题
腾讯云开发者
