我在尝试使用 pytest for FastAPI + Gino 应用程序运行测试时遇到一个奇怪的错误

Question

在测试中遇到奇怪的错误。我可能做错了什么，但我不知道到底是什么。

client = <starlette.testclient.TestClient object at 0x10b6a1400>

    @pytest.fixture
    @pytest.mark.anyio
    async def user(client):

        print('[-------->')
        # await init_db()

        # async with db.acquire():
>       user = await UserModel.create(email='test@gmail.com')

test_bug.py:81:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
env/lib/python3.9/site-packages/gino/crud.py:444: in _create_without_instance
    return await cls(**values)._create(bind=bind, timeout=timeout)
env/lib/python3.9/site-packages/gino/crud.py:477: in _create
    row = await bind.first(q)
env/lib/python3.9/site-packages/gino/engine.py:748: in first
    return await conn.first(clause, *multiparams, **params)
env/lib/python3.9/site-packages/asyncpg/pool.py:224: in release
    raise ex
env/lib/python3.9/site-packages/asyncpg/pool.py:214: in release
    await self._con.reset(timeout=budget)
env/lib/python3.9/site-packages/asyncpg/connection.py:1367: in reset
    await self.execute(reset_query, timeout=timeout)
env/lib/python3.9/site-packages/asyncpg/connection.py:318: in execute
    return await self._protocol.query(query, timeout)
asyncpg/protocol/protocol.pyx:323: in query
    ???
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _

>   ???
E   asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress

这是我的要求.txt

fastapi==0.70.0
gino==1.0.1
pytest==6.2.5
pytest-asyncio==0.16.0
requests==2.26.0

这是一个不起作用的独立示例。要查看错误，只需输入以下代码 test_bug.py 并运行 pytest。

import os
from typing import List
import pytest

from gino import Gino
from fastapi import APIRouter
from pydantic import BaseModel
from fastapi import FastAPI
from starlette.testclient import TestClient

router = APIRouter()

db = Gino()


async def init_db():
    await db.set_bind(os.environ['DATABASE_URL'])
    await db.gino.create_all()


class UserModel(db.Model):
    __tablename__ = 'user'

    id = db.Column(db.Integer(), primary_key=True)
    name = db.Column(db.Unicode())
    email = db.Column(db.Unicode(), unique=True, index=True)
    password_hash = db.Column(db.Unicode())


class UserSchema(BaseModel):
    id: int = 0
    name: str
    email: str
    password: str


class UserListSchema(BaseModel):
    objects: List[UserSchema]


@router.get("/users/", response_model=UserListSchema)
async def get_users():
    async with db.acquire():
        users = await UserModel.query.limit(200).gino.all()

    return UserListSchema.parse_obj({
        'objects': [x.to_dict() for x in users]
    })


def get_app():
    print('INIT APP')
    app = FastAPI(title="GINO FastAPI Demo")

    app.include_router(router, prefix='/API/v1')

    @app.on_event("startup")
    async def startup_event():
        print('Initialising DB')
        await init_db()
        print('DB was initialised')

    return app


@pytest.fixture
def client():

    with TestClient(get_app()) as client:
        yield client


@pytest.fixture
@pytest.mark.anyio
async def user(client):

    print('[-------->')
    # await init_db()

    # async with db.acquire():
    user = await UserModel.create(email='test@gmail.com')

    # async with db.acquire():
    users = await UserModel.query.limit(200).gino.all()
    print('.....=', user)
    print('....._', users)

    yield user


def test_users(user, client):
    response = client.get(
        "/API/v1/users",
        headers={},
    )
    print('=====', user, response.text)

    assert response.status_code == 200
    assert response.json() == {}

Answer 1

你可以将starlette的版本降级到0.14.2来解决这个问题