I am writing tests for a FastAPI application using httpx and pytest-asyncio. I want to ensure that the MongoDB database is cleared before each test method. However, despite setting up the fixture with scope="function"
and autouse=True
, it seems that the database is not being cleared before each test.
Here is the code for the tests and the fixture:
import pytest
from httpx import ASGITransport, AsyncClient
from motor.motor_asyncio import AsyncIOMotorClient
from app.main import app # Ensure your FastAPI app is correctly imported
@pytest.fixture(scope="function", autouse=True)
async def clear_db():
print("Clearing the database")
client = AsyncIOMotorClient("mongodb://localhost:27017")
await client["markdown_db"]["markdown_files"].delete_many({})
client.close()
@pytest.mark.asyncio
async def test_create_markdown():
transport = ASGITransport(app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
response = await client.post("/markdowns/", json={"content": "Company: TestCompanynContent goes here."})
assert response.status_code == 200, response.text
@pytest.mark.asyncio
async def test_read_markdown():
transport = ASGITransport(app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
create_response = await client.post("/markdowns/", json={"content": "Company: TestCompanynContent goes here."})
assert create_response.status_code == 200
markdown_id = create_response.json()["id"]
read_response = await client.get(f"/markdowns/{markdown_id}")
assert read_response.status_code == 200
data = read_response.json()
assert data["company"] == "TestCompany"
@pytest.mark.asyncio
async def test_update_markdown():
transport = ASGITransport(app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
create_response = await client.post("/markdowns/", json={"content": "Company: TestCompanynContent goes here."})
assert create_response.status_code == 200
markdown_id = create_response.json()["id"]
update_response = await client.put(f"/markdowns/{markdown_id}", json={"content": "Updated content."})
assert update_response.status_code == 200
data = update_response.json()
assert data["content"] == "Updated content."
@pytest.mark.asyncio
async def test_read_all_markdowns():
transport = ASGITransport(app)
async with AsyncClient(transport=transport, base_url="http://testserver") as client:
await client.post("/markdowns/", json={"content": "Company: TestCompany1nContent goes here."})
await client.post("/markdowns/", json={"content": "Company: TestCompany2nContent goes here."})
read_response = await client.get("/markdowns/")
assert read_response.status_code == 200
data = read_response.json()
assert len(data) >= 2
Why might the clear_db
fixture not be executing as expected before each test? Are there any best practices or corrections needed in the fixture or the test setup to ensure the database is cleared before each test method?