Why does my second test result changes depending on if i execute tests previously or not?

Question

I have the following piece of code:

conftest.py

fake_users = {
    "darth": User(
        id=1,
        first_name="Darth",
        last_name="Vader",
        email="[email protected]",
        password=User.hash_password("Anakin"),
        active=True,
        superuser=True,
    ),
    "luke": User(
        id=2,
        first_name="Luke",
        last_name="Skywalker",
        email="[email protected]",
        password=User.hash_password("Father"),
        active=True,
        superuser=True,
    ),
}


@pytest.fixture(scope="function")
def app_with_db():
    from app.main import app
    from app.api.db import SQLBaseModel, get_session
    engine = create_engine(
        settings.sql_alchemy_database_url,
        connect_args={"check_same_thread": False},
    )

    def override_get_session():
        with Session(engine, autoflush=True) as session:
            with session.begin():
                yield session


    app.dependency_overrides[get_session] = override_get_session

    SQLBaseModel.metadata.create_all(bind=engine)
    
    with Session(engine) as session:
        for user in fake_users.values():
            session.add(user)
        session.commit()
   
    yield app

    app.dependency_overrides = {}
    SQLBaseModel.metadata.drop_all(bind=engine)

then i have my tests.py

def test_create_user_success(app_with_db):
    user_data = {
        "email": "[email protected]",
        "password": "password",
        "first_name": "first",
        "last_name": "last",
    }
    headers = {"Content-Type": "application/json"}
    client = TestClient(app_with_db)
    create_user_response = client.post("/users/", json=user_data, headers=headers)
    print(create_user_response.json())
    assert create_user_response.status_code == 201
    assert create_user_response.json() is None


def test_create_user_already_exists(app_with_db):
    darth_user = fake_users["darth"]
    user_data = {
        "email": darth_user.email,
        "password": "password",
    }
    headers = {"Content-Type": "application/json"}
    client = TestClient(app_with_db)
    create_user_response = client.post("/users/", json=user_data, headers=headers)
    print(create_user_response.json())
    assert create_user_response.status_code == 418
    assert create_user_response.json()["detail"] == "Unhandled Error"

And I have a problem that if I execute the tests exactly as it is, my second test fails because it correctly creates the user via POST endpoint when it should fail because in the fixture they are created before yielding the app instance.

But if i comment the first test and execute the second one standalone it succeeds.

I understand that each test before being ran executes the code of the fixture until the yield and then once the test is finished it executes the piece of code after the yield. So the test should succeed in this case.

Also if I remove the SQLBaseModel.metadata.drop_all(bind=engine) sentence both tests run succesfully and i can see the db data as it should be:

Does anyone know what's going on here?

Answer 1

the problem is since requests are asynchronous the connection pool is one at a time so you have to setup event loop first. I used AsyncClient instead of TestClient and setting up event_loop in conftest.py file solved my problem still

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

@pytest_asyncio.fixture(scope="function")
async def client(app_with_db):
    async with AsyncClient(
            app=app_with_db, base_url="http://xxxx:7878", headers={"Content-Type": "application/json"},
    ) as c:
        yield c

then sample test_x.py

async def test_get_x(client):
    response = await client.get("/api/v1/x")
    res = response.json()
    assert response.status_code == 200
    assert isinstance(res, list)

I was just trying to test your scenario for non-Asynchronous conftest.py

import asyncio
from fastapi.testclient import TestClient
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.orm import DeclarativeBase

DATABASE_URL = "postgresql://...................."

class Base(DeclarativeBase):
    pass

@pytest.fixture(scope="session")
def main_app():
    from app.main import app
    engine = create_engine(
        DATABASE_URL,
            )
    Base.metadata.create_all(bind=engine)
    SessionLocal = sessionmaker(autoflush=False, bind=engine)
    def get_db():
        db = SessionLocal()
        try:
            yield db
        finally:
            db.close()
    app.dependency_overrides[get_db] = get_db
    yield app
    Base.metadata.drop_all(bind=engine)

# @pytest.fixture(scope="function")
# def client(main_app):
#     with TestClient(
#             app=main_app, base_url="http://localhost:1111", headers={"Content-Type": "application/json"},
#     ) as c:
#         yield c

test_item.py

from fastapi.testclient import TestClient
def test_create_item(main_app):
    item_data = {"name": "Test Item", "description": "Test description"}
    client = TestClient(main_app)
    response = client.post("/item", json=item_data)
    print(response.json())
    assert response.status_code == 200
def test_read_items3(main_app):
    client = TestClient(main_app)
    response = client.get("/items")
    assert response.status_code == 200

main.py

import time
from fastapi import Depends, FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Session

app = FastAPI()

DATABASE_URL = "postgresql://xxxxx"

engine = create_engine(DATABASE_URL)

SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# Base = declarative_base()
class Base(DeclarativeBase):
    pass

class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, index=True)
    description = Column(String, index=True)

@app.on_event("startup")
def startup() -> None: 
    Base.metadata.create_all(bind=engine)

class ItemCreate(BaseModel):
    name: str
    description: str = None

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/item")
def create_item(item: ItemCreate, db: Session = Depends(get_db)):
    db_item = Item(**item.dict())
    time.sleep(3)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

@app.get("/items")
def read_items(db: Session = Depends(get_db)):
    items = db.query(Item).all()
    return {"items": items}

this works well to me