How to run databases when doing testing

[dvrg]

Hi, I have question how to start run databases when doing testing ?

# conftest.py
SQLALCHEMY_DATABASE_URI = "sqlite:///./test.db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URI, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

class YOLOConnection(aiosqlite.Connection):
    def _get_reset_query(self):
        return None

database = databases.Database(
    SQLALCHEMY_DATABASE_URI, min_size=3, max_size=10, connection_class=YOLOConnection
)

Base.metadata.create_all(bind=engine)

@pytest.fixture(scope="function")
async def async_client() -> Generator:
    async with AsyncClient(
        app=app, base_url=f"{settings.SERVER_HOST}{settings.API_V1_STR}"
    ) as ac:
        yield ac

# test_device.py
@pytest.mark.anyio
async def test_get_multi(async_client: AsyncClient) -> None:
    devices = await crud_device.device.get_multi()
    assert devices

# models/device.py
class CRUDDevice(CRUDBase[Device, DeviceCreate, None]):
...
...
...
    async def get_multi(self, *, skip: int = 0, limit: int = 100) -> List[Device]:
        query = select(self.model).offset(skip).limit(limit)
        return await database.fetch_all(query=query)

device = CRUDDevice(Device)

When I run the test, error shown :

async def acquire(self) -> None:
        assert self._connection is None, "Connection is already acquired"
>       assert self._database._pool is not None, "DatabaseBackend is not running"
E       AssertionError: DatabaseBackend is not running

Thanks in advance!

[aminalaee]

I haven't run that locally (yet), but I guess you're missing a await database.connect() ? You create a Database instance but it's not awaited.

[aminalaee]

So why do you create another database in your conftest.py? I mean you can pass a different URI to the database.Database() but are you using something other than sqlite in your main and sqlite in test?

I got it working but removed min_size=3, max_size=10 as they're not accepted by sqlite.

[dvrg]

Yep. I'm using PostgreSQL in production mode and using SQLite in testing mode. I'm missing paste in error code, this is my update error code.

self = <databases.backends.postgres.PostgresConnection object at 0x11729f940>

    async def acquire(self) -> None:
        assert self._connection is None, "Connection is already acquired"
>       assert self._database._pool is not None, "DatabaseBackend is not running"
E       AssertionError: DatabaseBackend is not running

When start test, my application stay refer to PostgreSQL, not connection SQLite databases what I rewrite in conftest. I believe because the @app.on_event function in main.py call default databases.Database() what connect to PostgreSQL. My next questions are, how I create new connection to databases.Database() what I rewrite in conftest.py when doing testing ? Thank you

[aminalaee]

I'm not really sure how that would work out. Let's say you have a Decimal column in your models, that would be stored differently in PostgreSQL and SQLite.

You can have your Database setup in main application, and use env variables to set it up (if possible), and use different env variables for production and test.

# main.py

database = databases.Database(os.environ["DATABASE_URI"])

And set different env variables for production and development.

[dvrg]

Thank for your suggestions and answer.

[sumit-158]

I'm also stuck in a somewhat similar problem:

# conftest.py
import os

os.environ["ENV_STATE"] = "test"
from typing import AsyncGenerator, Generator
import pytest
from app.core.config import settings
from fastapi.testclient import TestClient
from httpx import AsyncClient, Request, Response
from app.models.user import users as user_table

from app.main import app
from app.db.session import database


@pytest.fixture(scope="session")
def anyio_backend():
    return "asyncio"


@pytest.fixture(autouse=True)
async def db() -> AsyncGenerator:
    await database.connect()
    yield database
    await database.disconnect()

@pytest.fixture()
async def client() -> Generator:
    yield TestClient(app)


@pytest.fixture()
async def async_client(client) -> AsyncGenerator:
    async with AsyncClient(app=app, base_url=client.base_url) as ac:
        yield ac
       
@pytest.fixture()
async def registered_user(async_client: AsyncClient) -> dict:
    user_details = {
        "first_name": "John",
        "last_name": "Doe",
        "email": "johndoe@example.in",
        "password": "pass123",
    }
    await async_client.post(f"{settings.API_V1_STR}/users/", json=user_details)
    query = user_table.select().where(user_table.c.email == user_details["email"])
    user = await database.fetch_one(query)
    user_details["id"] = user.id
    return user_details 
        

# test_user.py
from httpx import AsyncClient


async def test_create_user_endpoint(async_client: AsyncClient):
    user_data = UserCreate(
        first_name="John",
        last_name="Doe",
        email="johndoe@test.com",
        password="pass123",
    )
    response = await async_client.post(
        f"{settings.API_V1_STR}/users/", json=user_data.dict()
    )
    assert response.status_code == 200
    created_user = response.json()
    assert created_user["first_name"] == "John"
    assert created_user["last_name"] == "Doe"

async def test_already_registred_user(async_client: AsyncClient,registered_user: dict):
    user_data = UserCreate(
        first_name="John",
        last_name="Doe",
        email=registered_user["email"],
        password=registered_user["password"],
    )
    response = await async_client.post(
        f"{settings.API_V1_STR}/users/", json=user_data.dict()
    )
    assert response.status_code == 200
    created_user = response.json()
    assert created_user["first_name"] == "John"
    assert created_user["last_name"] == "Doe"
    

# config.py
from typing import Any, Dict
 
from pydantic import BaseSettings , PostgresDsn, validator


class GlobalConfig(BaseSettings):
    """
    Settings class for the application
    """
    ENV_STATE: str
    DATABASE_USER: str
    DATABASE_PASSWORD: str
    DATABASE_HOST: str
    DATABASE_PORT: int | str
    DATABASE_NAME: str
    DB_FORCE_ROLL_BACK: bool = False
    API_V1_STR: str = "/api/v1"
    SECRET_KEY: str
    TEXT_LOCAL_API_KEY: str
    TEXTLOCAL_SENDER: str
    TEXTLOCAL_TEMPLATE_ID: str
    SENDGRID_EMAIL: str
    SENDGRID_API_KEY: str
    ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24 * 8
    PROJECT_NAME: str = "Catalyst"
    DATABASE_URI: PostgresDsn | None = None

    @validator("DATABASE_URI", pre=True)
    def assemble_db_connection(cls, v: str | None, values: dict[str, Any]) -> Any:
        if isinstance(v, str):
            return v
        return PostgresDsn.build(
            scheme="postgresql",
            user=values.get("DATABASE_USER"),
            password=values.get("DATABASE_PASSWORD"),
            host=values.get("DATABASE_HOST"),
            port=str(values.get("DATABASE_PORT")),
            path=f"/{values.get('DATABASE_NAME') or ''}",
        )
    class Config:
        env_file = ".env"
        env_file_encoding = "utf-8"

class DevConfig(GlobalConfig):
    pass

class TestConfig(GlobalConfig):
    DB_FORCE_ROLL_BACK: bool = True


def get_config(env_state: str):
    """Instantiate config based on the environment."""
    configs = {"dev": DevConfig,  "test": TestConfig}
    return configs[env_state]()

settings = get_config(GlobalConfig().ENV_STATE)

#main.py
metadata.create_all(engine)


app = FastAPI(title="Catalyst")


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()
   

It is failing with this error:

 ERROR at teardown of test_create_user_endpoint 
 
   def finalizer() -> None:
       """Yield again, to finalize."""
   
       async def async_finalizer() -> None:
           try:
               await gen_obj.__anext__()
           except StopAsyncIteration:
               pass
           else:
               msg = "Async generator fixture didn't stop."
               msg += "Yield only once."
               raise ValueError(msg)

>       event_loop.run_until_complete(async_finalizer())

ERROR at setup of test_already_registred_user


def call_soon(self, callback, *args, context=None):
       """Arrange for a callback to be called as soon as possible.

       This operates as a FIFO queue: callbacks are called in the
       order in which they are registered.  Each callback will be
       called exactly once.

       Any positional arguments after the callback will be passed to
       the callback when it is called.
       """
>       self._check_closed()

Any help is much appreciated.