How to write tests?

[vladisa88]

Hi there! I want to know how can I use test database for testing fastapi app? When I repeat example code all my app is break. I think it happens because of double creation of the tables, once via migrations and other via metadata.create_all(engine).
When I create new engine with new database url and do something like this:

app.state.database = test_database
client = TestClient(app)

@pytest.fixture(autouse=True, scope="session")
def create_test_database():
    metadata.create_all(test_engine)
    yield
    metadata.drop_all(test_engine)

I get this error while running tests: AssertionError: DatabaseBackend is not running
Have anyone got an example of writing tests?

[collerek]

I depends if you are testing with client or without.

If you do only unit/logic testing this should work:

# fixture
@pytest.fixture(autouse=True, scope="module")  # adjust your scope
def create_test_database():
    engine = sqlalchemy.create_engine(DATABASE_URL)
    metadata.drop_all(engine) # i like to drop also before - even if test crash in the middle we start clean
    metadata.create_all(engine)
    yield
    metadata.drop_all(engine)

# actual test - note to test async you need pytest-asyncio and mark test as asyncio
@pytest.mark.asyncio
async def test_actual_logic():
    async with database:  # <= note this is the same database that used in ormar Models
        ... (logic)

If you want to test the actual app/api (fastapi) then use starlette TestClient -> note that this is only sync (no async version, at least last time I checked which was quite some time ago). To run async you can use httpx, although I never used it before, so you need to figure out the setup - I think fastapi docs have a section about this, in general it has api very similar to requests.

from fastapi import FastAPI
from starlette.testclient import TestClient

app = FastAPI()

# attach db to state
app.state.database = database

# define event like in production
@app.on_event("startup")
async def startup() -> None:
    database_ = app.state.database
    if not database_.is_connected:
        await database_.connect()


@app.on_event("shutdown")
async def shutdown() -> None:
    database_ = app.state.database
    if database_.is_connected:
        await database_.disconnect()


@pytest.fixture(scope="module")
def client():
    client = TestClient(app)
    yield client

# now the secret is that in order to trigger the events (and not have to manually connect the db)
# you have to use the client as context manager (so `with ...`).

def test_all_endpoints(): # note that with TestClient test is sync
    with client as client:  # this will connect/disconnect the db
        ... (logic)

Note that db creation/drop is needed in both cases, but didn't want to repeat the same code for clarity, example fixture in first snippet.

[vladisa88]

If we create a new FastAPI app instance, how it will test our current API? I will provide you an example of my code:

# test_somecode.py
import pytest
import databases
import sqlalchemy
from fastapi import FastAPI
from starlette.testclient import TestClient

from app import app

from conf.db import metadata, test_database, test_engine

app.state.database = test_database
# client = TestClient(app)

@pytest.fixture(autouse=True, scope="module")  # adjust your scope
def create_test_database():
    metadata.drop_all(test_engine) # i like to drop also before - even if test crash in the middle we start clean
    metadata.create_all(test_engine)
    yield
    metadata.drop_all(test_engine)

@app.on_event("startup")
async def startup() -> None:
    database_ = app.state.database
    if not database_.is_connected:
        await database_.connect()

@app.on_event("shutdown")
async def shutdown() -> None:
    database_ = app.state.database
    if database_.is_connected:
        await database_.disconnect()

@pytest.fixture(scope="module")
def client():
    client = TestClient(app)
    yield client

def test_populate_accounts(client):
    with client as client:
        response = client.post("/populate")
    assert response.status_code == 200

When I run this with pytest I get an error: AssertionError: DatabaseBackend is not running. And it related to PostgresBackend, but my test_database reference to sqlite database. I am really confused about it...

[collerek]

Yeah it was a dummy example.

Your actual application has to have the startup/shutdown events.
So add those to the app you import here:

from app import app # <= in you app or __init__ file

has to have those (defined in app)

@app.on_event("startup")
async def startup() -> None:
    database_ = app.state.database
    if not database_.is_connected:
        await database_.connect()

@app.on_event("shutdown")
async def shutdown() -> None:
    database_ = app.state.database
    if database_.is_connected:
        await database_.disconnect()

It also means that you have to import the database to you app.py file and set state there not in tests. Note that in every place you use database it has to be the same database, otherwise you connect different instance than the one later used.

When you use TestClient to test your up you just basically grab the production app and use it, only thing that should change is the database url. To switch configs best use environmental variables something like:

import os
env = os.environ.get('APPLICATION_ENV')  # name of the variable
if env == 'Development': # one or more configs
    DATABASE_URL = ...
    other_value = ...
else:
    DATABASE_URL = ...

Note that now you have to set this variable, it's operating system specific so best to google if you don't know how. Saw that you already have conf file so you can switch multiple variables depending on env var value.

If after this you will stil have problems I can help you with that if you post link to your repo (if it's public).

[vladisa88]

Thank you so much! It helped as guessed. You are great!

[pawamoy]

I'll add another level of complexity: what if I want to run my tests in parallel, for example with the pytest-xdist plugin? Can I use the same database? Maybe if I wrap every test in a transaction there will be no race condition? What if a route called by the test client also starts a transaction? Do nested transaction work? What about FastAPI background tasks?

At some point I think I should really use one unique database per test, but I'm having trouble mocking it in pytest. Should I mock it in BaseMeta? Will all the models use the mocked version?

[pawamoy]

I got something working. Will post the solution in a few days.

[collerek]

Yes if you mock it in base all of them should use the same (if you use the same base for all models that is 😆 ).

You can spin transactions, not sure if nested transactions are supported, though.

Let me know what you got and if I can help somehow.

[dbatten5]

I got something working. Will post the solution in a few days.

i'm interested in your solution to this - did you post it anywhere?

[pawamoy]

Not yet, I've procrastinated a lot 😄
Working on it right now 🙂
I'll probably publish it this coming week.

[pawamoy]

And I've procrastinated a lot more. But here it is now! https://pawamoy.github.io/posts/testing-fastapi-ormar-alembic-apps/

[5h44n]

Building on this -- in local development I have a single postgres database. I want to use that database for running unit tests, as well as for holding local data as I develop. Is there a way to preserve my local data but also ensure I get a fresh database each time for a given unit test? Maybe using databases.Database(db_url, force_rollback=True) or transactions? Or do I need to create a separate local postgres database solely for testing?

Here's my current setup which uses a SQLite in memory test db.

main.py:

# ...
def create_application(database: databases.Database = None):
    application = FastAPI()

    # register routers
    application.include_router(examples.router)

    # register exception handlers
    application.add_exception_handler(ExampleException, example_exception_handler)
  
    # set app state
    application.state.database = database

    return application

database = databases.Database("postgresql://postgres@drivedigital_db:5432/postgres")
app = create_application(database=database)


@app.on_event("startup")
async def startup():
    # initialize database
    database_ = app.state.database
    if not database_.is_connected:
        await database_.connect()


@app.on_event("shutdown")
async def shutdown():
    # disconnect from database
    database_ = app.state.database
    if database_.is_connected:
        await database_.disconnect()

conftest.py:

# ...
from httpx import AsyncClient

from .main import startup, shutdown

metadata = sqlalchemy.MetaData()
database = databases.Database("sqlite:///:memory")

# this actually lives in a db.py file but putting it here for clarity
class BaseMeta(ormar.ModelMeta):
    metadata = metadata
    database = database


@pytest.fixture
def db():
    engine = sqlalchemy.create_engine("sqlite:///:memory")
    BaseMeta.metadata.drop_all(bind=engine)
    BaseMeta.metadata.create_all(bind=engine)
    yield database
    BaseMeta.metadata.drop_all(bind=engine)


@pytest.fixture
def app_with_db(db):
    app = create_application(database=db)
    app.on_startup = [startup]
    app.on_shutdown = [shutdown]
    yield app


@pytest.fixture
def async_client_with_db(app_with_db):
    yield AsyncClient(app=app_with_db, base_url="http://test")

BTW: thanks for creating ormar, its been a real bright spot in the world of asyncio web development 🙏🏾

[collerek]

Hi, although you can rollback and force rollback transactions there is always a risk that some unhandled exceptions, power outage or whatever causes data corruption and your db can loose integrity or have some corrupted data. I.e. there is bug in databases that causes in some specific situation that transaction is not rolled back.

Therefore I recommend setting a new db for tests. In your case I would setup a docker image that copies data from your existing database for tests and is removed after test runs. Note the word copies, not mounts as if you mount your host file system you would affect the original database.

You can follow a guide similar to https://stackoverflow.com/questions/57770303/how-to-copy-and-use-existing-postgres-data-folder-into-docker-postgres-container

Apart from docker and rollback a third option is to write a custom script that cleans the database to initial state on teardown of test suite. This is usually done either based on users (test run as different user) or timestamps so you need this information in your tables. This strategy is often used i.e. in integration testing when you cannot rollback directly after db calls as you need the data for later checks. This can be cumbersome however, as you either need to delete the data in proper order (children before parents to not violate foreign keys) or setup cascade deletes in your db schema definition to delete children on deletion of the parent.

Thanks for the kind words! :)

[5h44n]

Thanks for this explanation, that all makes sense, especially after seeing the issues databases is having with transactions. I want to add some clarification here though for people who might come across this: I was not interested in explicitly using the same local postgres database as some sort of setup for my testing. Rather, since I'm developing against a postgres database, I also just wanted my unit tests to run against any postgres database for consistency's sake (and testing my ability to handle database specific exceptions, etc.)

Therefore the solution was much easier following this SO post: I just added a second postgres database container in my local docker-compose.yml running on port 5433 and pointed my pytest db fixture at it.

[ponytailer]

In tests, use the test-containers to define the uint test db backend.

In my flask application tests: 

@pytest.fixture(scope="session")
def db_init():
    image = "xxxx"
    with MySqlContainer(image) as mysql:
        app.config["TESTING"] = True
        app.config["SQLALCHEMY_DATABASE_URI"] = mysql.get_connection_url() 
        with app.app_context():
            db.create_all()
        yield mysql

[slavpetroff]

This is really stale info before the major rework in version 0.20+. Could you please update the examples how to setup properly tests using ormar and pytest? @collerek

P.S.

Also would be really helpful to showcase usage of inmemory DB, because thats the most commonly used approach in test environment.