我想使用两种不同的 httpx 客户端，一种与实际数据库交互，另一种与测试数据库交互。但由于某种原因，所有事务都发生在实际数据库上。
# conftest.py
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="function")
async def real_client():
async with LifespanManager(app):
async with AsyncClient(
transport=ASGITransport(app), base_url="http://localhost"
) as c:
yield c
async def init_db(db_url, create_db: bool = False, schemas: bool = False) -> None:
"""Initial database connection"""
await Tortoise.init(
db_url=db_url, modules={"models": ["app.database.models"]}, _create_db=create_db
)
if create_db:
print(f"Database created! {db_url = }")
if schemas:
await Tortoise.generate_schemas()
print("Success to generate schemas")
async def init():
DB_URL = "sqlite://:memory:"
await init_db(DB_URL, True, True)
dashboard_permission_model = CreatePermission(
name="Dashboard", description="Dashboard Page"
)
camera_permission_model = CreatePermission(name="Camera", description="Camera Page")
await Permission.create(**dashboard_permission_model.dict())
await Permission.create(**camera_permission_model.dict())
admin_role_model = CreateRole(name="Admin", description="Admin dashboard")
user = await User.create(
first_name="Admin",
last_name="",
email=ADMIN_EMAIL,
password=get_password_hash(password="some@password"),
)
admin_role = await Role.create(**admin_role_model.dict())
await UserRole.create(role=admin_role, user=user)
# Add all permission to admin
all_permissions = await Permission.all()
for permission in all_permissions:
await UserPermission.create(permission=permission, user=user)
@pytest.fixture(scope="function")
async def test_client():
async with AsyncClient(app=ASGITransport(app), base_url="http://test") as client:
yield client
@pytest.fixture(scope="session", autouse=True)
async def initialize_tests():
await init()
yield
await Tortoise._drop_databases()
下面使用测试客户端（测试数据库）测试角色路由器：
# test_role.py
@pytest.mark.anyio
async def test_creat_role(
test_client: AsyncClient, auth_headers: dict[str, str], role_data
):
response = await test_client.post(
url="/role/add", headers=auth_headers, json=role_data
)
logging.debug(response)
assert response.status_code == 200
assert response.json()["name"] == "Test role"
assert response.json()["description"] == "Some test role"
下面测试的是使用真实客户端（实际数据库）的图形路由器：
# test_graph.py
@pytest.mark.anyio
@pytest.mark.parametrize(
"camera_ids, start_time, end_time, location_ids, time_frame",
[
([-8, -10], "2024-04-01", "2024-04-01", [], "day"), # Invalid camera ids
],
)
async def test_dwell_time_and_trends_invalid_camera_ids(
real_client: AsyncClient,
auth_headers: dict[str, str],
current_user: int,
camera_ids: list[int],
start_time: str,
end_time: str,
location_ids: list[int],
time_frame: str,
):
store_ids = await get_store_ids(current_user)
response = await real_client.get(
url="/dwell-time-and-trends",
headers=auth_headers,
params={
"store_ids": store_ids,
"camera_ids": camera_ids,
"start_time": start_time,
"end_time": end_time,
"location_ids": location_ids,
"time_frame": time_frame,
},
)
logging.debug(response.content)
assert response.status_code == 500
assert response.json()["detail"] == "list index out of range"
当我忽略使用 real_client 固定装置的测试用例时，它会起作用，其中创建了本地数据库并且所有事务都发生在其上。但是当我运行所有测试时，所有事务都发生在实际数据库上。
我有一些理论，可能与anyio后端有关，或者客户端在同一会话中发生冲突。我观察到，首先调用的客户端是持久的，并且在整个会话期间使用。
您可以尝试将 real_client 的范围从会话更改为模块以及当前用户情况的相关固定装置等。并将initialize_test()固定装置与 test_client 固定装置合并将确保测试数据库相关事务仅在使用 test_client 时发生.
确保 real_client 在模块级别运行将确保只有您所需的模块才会使用 real_client，而测试的其余部分可以在整个会话中使用 test_client。
你的conftest.py
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="module")
async def real_client():
async with LifespanManager(app):
async with AsyncClient(
transport=ASGITransport(app), base_url="http://localhost"
) as c:
yield c
@pytest.fixture(scope="module")
async def user_email():
return await User.filter(id=1).first()
@pytest.fixture(scope="module")
async def current_user(user_email):
return await User.filter(email=user_email).first()
@pytest.fixture(scope="module")
def auth_headers(user_email):
refresh_token = create_refresh_token(user_email)
return {"Authorization": f"Bearer {refresh_token}"}
async def init_db(db_url, create_db: bool = False, schemas: bool = False) -> None:
"""Initial database connection"""
await Tortoise.init(
db_url=db_url, modules={"models": ["app.database.models"]}, _create_db=create_db
)
if create_db:
print(f"Database created! {db_url = }")
if schemas:
await Tortoise.generate_schemas()
print("Success to generate schemas")
async def init():
DB_URL = "sqlite://:memory:"
await init_db(DB_URL, True, True)
dashboard_permission_model = CreatePermission(
name="Dashboard", description="Dashboard Page"
)
camera_permission_model = CreatePermission(name="Camera", description="Camera Page")
await Permission.create(**dashboard_permission_model.dict())
await Permission.create(**camera_permission_model.dict())
admin_role_model = CreateRole(name="Admin", description="Admin dashboard")
user = await User.create(
first_name="Admin",
last_name="",
email=ADMIN_EMAIL,
password=get_password_hash(password="some@password"),
)
admin_role = await Role.create(**admin_role_model.dict())
await UserRole.create(role=admin_role, user=user)
# Add all permission to admin
all_permissions = await Permission.all()
for permission in all_permissions:
await UserPermission.create(permission=permission, user=user)
@pytest.fixture(scope="session")
async def test_client():
await init()
async with AsyncClient(transport=ASGITransport(app), base_url="http://test") as client:
yield client
await Tortoise._drop_databases()
希望这对您有用！