I want to use two different httpx clients one which interacts with the actual database and other which interact with test database. But for some reason all the transactions happen on the actual database.
# 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()
Below test's the role router using the test client(test database):
# 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"
Below test's the graph router using the real client(actual database):
# 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"
It works when I ignore the test cases which use the real_client fixture where a local db is created and all the transactions happen on it. But when I run all the tests then all the transactions happen on the actual database.
I have some theory maybe it's related to the anyio backend or the clients are conflicting in the same session. I have observed that whichever client is called first is persistent and is used throughout the session.
You can try changing the scope of the real_client from session to module and related fixtures for your case the current_user, etc. And merging the initialize_test() fixture with the test_client fixture will make sure the test database related transactions only happen while using the test_client.
Making sure real_client runs at module level will ensure only your required modules will use the real_client and rest of the test can use the test_client for the complete session.
Your 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()
Hope this works for you!