我有以下代码:
conftest.py
fake_users = {
"darth": User(
id=1,
first_name="Darth",
last_name="Vader",
email="[email protected]",
password=User.hash_password("Anakin"),
active=True,
superuser=True,
),
"luke": User(
id=2,
first_name="Luke",
last_name="Skywalker",
email="[email protected]",
password=User.hash_password("Father"),
active=True,
superuser=True,
),
}
@pytest.fixture(scope="function")
def app_with_db():
from app.main import app
from app.api.db import SQLBaseModel, get_session
engine = create_engine(
settings.sql_alchemy_database_url,
connect_args={"check_same_thread": False},
)
def override_get_session():
with Session(engine, autoflush=True) as session:
with session.begin():
yield session
app.dependency_overrides[get_session] = override_get_session
SQLBaseModel.metadata.create_all(bind=engine)
with Session(engine) as session:
for user in fake_users.values():
session.add(user)
session.commit()
yield app
app.dependency_overrides = {}
SQLBaseModel.metadata.drop_all(bind=engine)
然后我有我的测试。py
def test_create_user_success(app_with_db):
user_data = {
"email": "[email protected]",
"password": "password",
"first_name": "first",
"last_name": "last",
}
headers = {"Content-Type": "application/json"}
client = TestClient(app_with_db)
create_user_response = client.post("/users/", json=user_data, headers=headers)
print(create_user_response.json())
assert create_user_response.status_code == 201
assert create_user_response.json() is None
def test_create_user_already_exists(app_with_db):
darth_user = fake_users["darth"]
user_data = {
"email": darth_user.email,
"password": "password",
}
headers = {"Content-Type": "application/json"}
client = TestClient(app_with_db)
create_user_response = client.post("/users/", json=user_data, headers=headers)
print(create_user_response.json())
assert create_user_response.status_code == 418
assert create_user_response.json()["detail"] == "Unhandled Error"
我有一个问题,如果我完全按照原样执行测试,我的第二个测试会失败,因为它在失败时通过 POST 端点正确创建了用户,因为在固定装置中它们是在生成应用程序实例之前创建的。
但是如果我评论第一个测试并独立执行第二个测试,它就会成功。
我理解,每个测试在运行之前都会执行夹具的代码,直到屈服,然后一旦测试完成,它就会执行屈服之后的代码段。所以在这种情况下测试应该成功。
此外,如果我删除
SQLBaseModel.metadata.drop_all(bind=engine)
句子,两个测试都会成功运行,并且我可以看到应有的数据库数据:
有人知道这是怎么回事吗?
问题是由于请求是异步的,因此连接池一次只有一个,因此您必须首先设置事件循环。 我使用 AsyncClient 而不是 TestClient 并在 conftest.py 文件中设置 event_loop 仍然解决了我的问题
@pytest_asyncio.fixture(scope="session")
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest_asyncio.fixture(scope="function")
async def client(app_with_db):
async with AsyncClient(
app=app_with_db, base_url="http://xxxx:7878", headers={"Content-Type": "application/json"},
) as c:
yield c
然后采样 test_x.py
async def test_get_x(client):
response = await client.get("/api/v1/x")
res = response.json()
assert response.status_code == 200
assert isinstance(res, list)