Skip to content

Test Client

I'm sure that you already faced the problem with testing your database anmd thinking about a way of making sure the tests against models would land in a specific targeted database instead of the one used for development, right?

Well, at least I did and it is annoying the amount of setup required to make it happen and for that reason, Saffier provides you already one client that exctly that job for you.

For this example, we will be using Saffier ORM as the author is the same and helps with the examples.

Before continuing, make sure you have the databasez test client installed with the needed requirements.

$ pip install databasez[testing]

History behind it

There are a lot of frameworks and helpers for databases that can help you with the test_ database creation. A great example is Django that does it automatically for you when running the unit tests.

When this was initially thought, sqlalchemy_utils was the package used to help with the process of creation and dropping of the databases. the issue found was the fact that up to the version 0.40.0, the package wasn't updated for a very long time and no async support was added.

When DatabaseTestClient was created was with that in mind, by rewritting some of those functionalities but with async support and native to the databasez.

DatabaseTestClient

This is the client you have been waiting for. This object does a lot of magic for you and will help you manage those stubborn tests that should land on a test_ database.

from databasez.testclient import DatabaseTestClient

Parameters

  • url - The database url for your database. This can be in a string format or in a databasez.DatabaseURL.

    from databasez import DatabaseURL
    
  • force_rollback - This will ensure that all database connections are run within a transaction that rollbacks once the database is disconnected.

    Default: False

  • use_existing - Uses the existing test_ database if previously created and not dropped.

    Default: False

  • drop_database - Ensures that after the tests, the database is dropped.

    Default: False

How to use it

This is the easiest part because is already very familiar with the Database used by Saffier. In fact, this is an extension of that same object with a lot of testing flavours.

Let us assume you have a database url like this following:

DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/my_db"

We know the database is called my_db, right?

When using the DatabaseTestClient, the client will ensure the tests will land on a test_my_db.

Pretty cool, right?

Nothing like an example to see it in action.

tests.py
import datetime
import decimal
import ipaddress
import uuid
from enum import Enum

import pytest
import saffier
from databasez.testclient import DatabaseTestClient
from saffier.db.models import fields

from tests.settings import DATABASE_URL

database = DatabaseTestClient(DATABASE_URL, drop_database=True)
models = saffier.Registry(database=database)

pytestmark = pytest.mark.anyio


def time():
    return datetime.datetime.now().time()


class StatusEnum(Enum):
    DRAFT = "Draft"
    RELEASED = "Released"


class Product(saffier.Model):
    id = fields.IntegerField(primary_key=True)
    uuid = fields.UUIDField(null=True)
    created = fields.DateTimeField(default=datetime.datetime.now)
    created_day = fields.DateField(default=datetime.date.today)
    created_time = fields.TimeField(default=time)
    created_date = fields.DateField(auto_now_add=True)
    created_datetime = fields.DateTimeField(auto_now_add=True)
    updated_datetime = fields.DateTimeField(auto_now=True)
    updated_date = fields.DateField(auto_now=True)
    data = fields.JSONField(default={})
    description = fields.CharField(blank=True, max_length=255)
    huge_number = fields.BigIntegerField(default=0)
    price = fields.DecimalField(max_digits=5, decimal_places=2, null=True)
    status = fields.ChoiceField(StatusEnum, default=StatusEnum.DRAFT)
    value = fields.FloatField(null=True)

    class Meta:
        registry = models


class User(saffier.Model):
    id = fields.UUIDField(primary_key=True, default=uuid.uuid4)
    name = fields.CharField(null=True, max_length=16)
    email = fields.EmailField(null=True, max_length=256)
    ipaddress = fields.IPAddressField(null=True)
    url = fields.URLField(null=True, max_length=2048)
    password = fields.PasswordField(null=True, max_length=255)

    class Meta:
        registry = models


class Customer(saffier.Model):
    name = fields.CharField(null=True, max_length=16)

    class Meta:
        registry = models


@pytest.fixture(autouse=True, scope="module")
async def create_test_database():
    await models.create_all()
    yield
    await models.drop_all()


@pytest.fixture(autouse=True)
async def rollback_transactions():
    with database.force_rollback():
        async with database:
            yield


async def test_model_crud():
    product = await Product.query.create()
    product = await Product.query.get(pk=product.pk)
    assert product.created.year == datetime.datetime.now().year
    assert product.created_day == datetime.date.today()
    assert product.created_date == datetime.date.today()
    assert product.created_datetime.date() == datetime.datetime.now().date()
    assert product.updated_date == datetime.date.today()
    assert product.updated_datetime.date() == datetime.datetime.now().date()
    assert product.data == {}
    assert product.description == ""
    assert product.huge_number == 0
    assert product.price is None
    assert product.status == StatusEnum.DRAFT
    assert product.value is None
    assert product.uuid is None

    await product.update(
        data={"foo": 123},
        value=123.456,
        status=StatusEnum.RELEASED,
        price=decimal.Decimal("999.99"),
        uuid=uuid.UUID("f4e87646-bafa-431e-a0cb-e84f2fcf6b55"),
    )

    product = await Product.query.get()
    assert product.value == 123.456
    assert product.data == {"foo": 123}
    assert product.status == StatusEnum.RELEASED
    assert product.price == decimal.Decimal("999.99")
    assert product.uuid == uuid.UUID("f4e87646-bafa-431e-a0cb-e84f2fcf6b55")

    last_updated_datetime = product.updated_datetime
    last_updated_date = product.updated_date
    user = await User.query.create()
    assert isinstance(user.pk, uuid.UUID)

    user = await User.query.get()
    assert user.email is None
    assert user.ipaddress is None
    assert user.url is None

    await user.update(
        ipaddress="192.168.1.1",
        name="Test",
        email="test@saffier.com",
        url="https://saffier.com",
        password="12345",
    )

    user = await User.query.get()
    assert isinstance(user.ipaddress, (ipaddress.IPv4Address, ipaddress.IPv6Address))
    assert user.password == "12345"

    assert user.url == "https://saffier.com"
    await product.update(data={"foo": 1234})
    assert product.updated_datetime != last_updated_datetime
    assert product.updated_date == last_updated_date

What is happening

Well, this is rather complex test and actually a real one from Saffier and what you can see is that is using the DatabaseTestClient which means the tests against models, fields or whatever database operation you want will be on a test_ database.

But you can see a drop_database=True, so what is that?

Well drop_database=True means that by the end of the tests finish running, drops the database into oblivion.