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. It supports the same types like normal Database objects and has a special handling for subclasses of DatabaseTestClient.

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

    Default: None, copy default or testclient_default_force_rollback (defaults to False)

  • full_isolation - Special mode for using force_rollback with nested queries. This parameter fully isolates the global connection in an extra thread. This way it is possible to use blocking operations like locks with force_rollback. This parameter has no use when used without force_rollback and causes a slightly slower setup (Lock is initialized). It is required for edgy or other frameworks which use threads in tests and the force_rollback parameter. For the DatabaseTestClient it is enabled by default.

    Default: None, copy default or testclient_default_full_isolation (defaults to True)

  • poll_interval - When using multithreading, the poll_interval is used to retrieve results from other loops. It defaults to a sane value.

    Default: None, copy default or testclient_default_poll_interval

  • lazy_setup - This sets up the db first up on connect not in init.

    Default: None, True if copying a database or testclient_default_lazy_setup (defaults to False)

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

    Default: testclient_default_use_existing (defaults to False)

  • drop_database - Ensures that after the tests, the database is dropped. The corresponding attribute is drop. When the setup fails, it is automatically set to False.

    Default: testclient_default_drop_database (defaults to False)

  • test_prefix - Allow a custom test prefix or leave empty to use the url instead without changes.

    Default: testclient_default_test_prefix (defaults to test_)

Subclassing

The defaults of all parameters except the url can be changed by providing in a subclass a different value for the attribute:

testclient_default_<parameter name>

There are also 2 knobs for the operation timeout (setting up DB, dropping databases):

testclient_operation_timeout

Default: 4.

and the limit

testclient_operation_timeout_init for the non-lazy setup in init of the database.

Default: 8.

How to use it

This is the easiest part because is already very familiar with the Database used by Edgy or 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 saffier.db.models import fields

from databasez.testclient import DatabaseTestClient
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.