[Python] I don't understand why my test database is not being cleared and how to fix it

Discussão em 'Python' iniciado por Stack, Julho 29, 2025.

  1. Stack

    Stack Membro Participativo

    I'm trying to write integration tests for my FastAPI registration endpoint using SQLModel and an async PostgreSQL test database but after succesful register database don't cleaned up even i have

    await conn.run_sync(SQLModel.metadata.drop_all)


    My current code

    conftest.py

    import pytest
    import asyncio
    from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
    from sqlmodel import SQLModel
    from httpx import AsyncClient
    from main import app
    from backend.app.database.database import get_session

    TEST_DATABASE_URL = "postgresql+asyncpg://test_user:test_pass@localhost:5432/test_db"

    engine_test = create_async_engine(TEST_DATABASE_URL, echo=True)
    TestingSessionLocal = async_sessionmaker(
    bind=engine_test, expire_on_commit=False, class_=AsyncSession
    )


    async def override_get_session():
    async with TestingSessionLocal() as db:
    yield db


    app.dependency_overrides[get_session] = override_get_session


    @pytest.fixture(scope="session")
    async def setup_database():
    async with engine_test.begin() as conn:
    await conn.run_sync(SQLModel.metadata.drop_all) # Очищаем после тестов

    await conn.run_sync(SQLModel.metadata.create_all) # Создаем таблицы
    yield
    async with engine_test.begin() as conn:
    await conn.run_sync(SQLModel.metadata.drop_all) # Очищаем после тестов


    @pytest.fixture(scope="module")
    async def client():
    async with AsyncClient(app=app, base_url="http://test") as c:
    yield c


    @pytest.fixture(scope="session")
    def event_loop():
    loop = asyncio.get_event_loop_policy().new_event_loop()
    yield loop
    loop.close()



    test_auth.py

    import pytest
    from httpx import AsyncClient


    @pytest.mark.asyncio
    async def test_register_user(
    client: AsyncClient,
    ):
    response = await client.post(
    "/api/v01/register/",
    data={
    "username": "testusername",
    "email": "test@example.com",
    "password": "strongpassword",
    "role": "reader",
    },
    )

    assert response.status_code == 302
    assert response.headers["location"] == "/profile/"



    I'm using Python 3.8.10.

    I tried many different options with chatgpt but it's all time don't work or giving me errors like "RuntimeError: Task <Task pending name='Task-4' coro=<test_register_user() running at c:\ "

    Continue reading...

Compartilhe esta Página