Я хочу использовать два разных 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 или конфликтуют клиенты в одном сеансе. Я заметил, что тот клиент, который вызывается первым, является постоянным и используется на протяжении всего сеанса.
@volkan, я обновил описание примерами тестов как для реального, так и для тестового клиента. Кстати, для вашей информации я использую черепаху-орм.






Вы можете попробовать изменить область действия real_client с сеанса на модуль и связанные с ним приспособления для вашего случая, current_user и т. д. А объединение приспособления 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()
Надеюсь, что это работает для вас!
Можете ли вы также поделиться своими тестовыми кодами? Спасибо