Testing with database state

Many times, for testing, we need to have data in a database to tests how function behave on them. This is usually done by setting up an application database with fixtures.

The main problem of that is that when we alter the database schema, fixtures must be updated. On the other way, if the database state is based on passed commands, then, this problem disappear.

Now lets update our tests in order to initialize a database state using fixtures.

So lets add this fixture in our main conftest file (tests/conftest.py).

import pytest
from reading_club.service.uow import AbstractUnitOfWork

from messagebus import AsyncMessageBus


@pytest.fixture
async def uow_with_data(
    uow: AbstractUnitOfWork, bus: AsyncMessageBus, params
) -> AbstractUnitOfWork:
    async with uow as transaction:
        for command in params.get("commands", []):
            await bus.handle(command, transaction)
        await transaction.commit()
    return uow

This fixture is dead simple, it is a parametrized fixtures, that consume a parameter params, which is a dict that contains a list of commands to be play by the bus.

This initialize the database using all the registered service handler, without any database details. Those service handler must be properly tests rights, to get the proper database state but you are probably aware of that.

Not that this fixture does not revert its changes after the test execution. This is not necessary because we create a new state from scratch on every tests in order to ensure that every tests are fully isolated.

lets rewrite and parametrized our tests that are should.

from collections.abc import Mapping
from typing import Any

import pytest
from lastuuid.dummies import uuidgen
from reading_club.adapters.uow_sqla.uow import SQLUnitOfWork
from reading_club.domain.messages import RegisterBook
from reading_club.domain.model import Book
from reading_club.service.repositories import BookRepositoryError
from result import Err, Ok


@pytest.mark.parametrize(
    "params",
    [
        {
            "commands": [
                RegisterBook(
                    id=uuidgen(),
                    title="Domain Driven Design",
                    author="Eric Evans",
                    isbn="0-321-12521-5",
                )
            ]
        }
    ],
)
async def test_book_add_err(
    params: Mapping[str, Any], uow_with_data: SQLUnitOfWork, book: Book
):
    uow = uow_with_data
    book.id = params["commands"][0].id
    async with uow as transaction:
        res = await uow.books.add(book)
        assert res.is_err()
        assert res.unwrap_err() == BookRepositoryError.integrity_error
        await transaction.rollback()

    # Since it does not work, the bus can't see the book messages.
    assert uow.books.seen == []


@pytest.mark.parametrize(
    "params",
    [
        pytest.param(
            {
                "book_id": uuidgen(1),
                "commands": [
                    RegisterBook(
                        id=uuidgen(1),
                        title="Domain Driven Design",
                        author="Eric Evans",
                        isbn="0-321-12521-5",
                    ),
                    RegisterBook(
                        id=uuidgen(2),
                        title="Architecture Patterns With Python",
                        author="Harry Percival and Bob Gregory",
                        isbn="978-1492052203",
                    ),
                ],
                "expected_result": Ok(
                    Book(
                        id=uuidgen(1),
                        title="Domain Driven Design",
                        author="Eric Evans",
                        isbn="0-321-12521-5",
                    )
                ),
            },
            id="return a known book",
        ),
        pytest.param(
            {
                "book_id": uuidgen(),
                "commands": [
                    RegisterBook(
                        id=uuidgen(),
                        title="Domain Driven Design",
                        author="Eric Evans",
                        isbn="0-321-12521-5",
                    ),
                    RegisterBook(
                        id=uuidgen(),
                        title="Architecture Patterns With Python",
                        author="Harry Percival and Bob Gregory",
                        isbn="978-1492052203",
                    ),
                ],
                "expected_result": Err(BookRepositoryError.not_found),
            },
            id="return an error",
        ),
    ],
)
async def test_book_by_id(params: Mapping[str, Any], uow_with_data: SQLUnitOfWork):
    uow = uow_with_data
    # Now, tests that the book is here
    async with uow as transaction:
        res = await uow.books.by_id(params["book_id"])
        await transaction.rollback()

    assert res == params["expected_result"]

Now lets run our tests

$ poetry run pytest -sxv
========================== test session starts ==========================
collected 9 items

tests/test_service_handler_add_book.py::test_register_book[params0] PASSED
tests/test_service_handler_add_book.py::test_bus_handler PASSED
tests/uow_sqla/test_repositories.py::test_book_add_ok PASSED
tests/uow_sqla/test_repositories.py::test_book_add_err[params0] PASSED
tests/uow_sqla/test_repositories.py::test_book_by_id[return a known book] PASSED
tests/uow_sqla/test_repositories.py::test_book_by_id[return an error] PASSED
tests/uow_sqla/test_repositories.py::test_eventstore_add PASSED
tests/uow_sqla/test_transaction.py::test_commit PASSED
tests/uow_sqla/test_transaction.py::test_rollback PASSED
=========================== 9 passed in 0.48s ===========================

Note

You may notice that the tests run slower than before, which was predictible. But this is quite acceptable. By the way, using an in memory unit of work and the time saved by having maintainable fixture is what matter the most.

Now, lets rewrite another test from the chapter 4 where we have two tests in one. We have our new fixture that could be used here to get the database state and get our two test runned sequently.

Here is the context, from the file test_service_handler_add_book.py

from lastuuid.dummies import uuidgen
from reading_club.domain.messages import RegisterBook
from reading_club.domain.model import Book
from reading_club.service.handlers.book import register_book
from reading_club.service.repositories import BookRepositoryError
from reading_club.service.uow import AbstractUnitOfWork


async def test_register_book(register_book_cmd: RegisterBook, uow: AbstractUnitOfWork):
    async with uow as transaction:
        operation = await register_book(register_book_cmd, transaction)
        assert operation.is_ok()
        book = await transaction.books.by_id(register_book_cmd.id)
        assert book.is_ok()
        assert book.unwrap() == Book(
            id=uuidgen(2),
            title="Domain Driven Design",
            author="Eric Evans",
            isbn="0-321-12521-5",
        )
        await transaction.commit()

    async with uow as transaction:
        operation = await register_book(register_book_cmd, transaction)
        assert operation.is_err()
        assert operation.unwrap_err() == BookRepositoryError.integrity_error
        await transaction.rollback()

Lets rewrite it with our new fixture:

from collections.abc import Mapping
from typing import Any

import pytest
from lastuuid.dummies import uuidgen
from reading_club.domain.messages import BookRegistered, RegisterBook
from reading_club.domain.model import Book
from reading_club.service.handlers.book import register_book
from reading_club.service.repositories import BookRepositoryError
from reading_club.service.uow import AbstractUnitOfWork
from result import Err, Ok


@pytest.mark.parametrize(
    "params",
    [
        pytest.param(
            {
                "expected_result": Ok(...),
                "expected_book": Ok(
                    Book(
                        id=uuidgen(1),
                        title="Domain Driven Design",
                        author="Eric Evans",
                        isbn="0-321-12521-5",
                    )
                ),
                "expected_messages": [
                    BookRegistered(
                        id=uuidgen(1),
                        isbn="0-321-12521-5",
                        title="Domain Driven Design",
                        author="Eric Evans",
                    )
                ],
            },
            id="ok",
        ),
        pytest.param(
            {
                "commands": [
                    RegisterBook(
                        id=uuidgen(1),
                        title="Architecture Patterns With Python",
                        author="Harry Percival and Bob Gregory",
                        isbn="978-1492052203",
                    )
                ],
                "expected_result": Err(BookRepositoryError.integrity_error),
                "expected_book": Ok(
                    Book(
                        id=uuidgen(1),
                        title="Architecture Patterns With Python",
                        author="Harry Percival and Bob Gregory",
                        isbn="978-1492052203",
                    )
                ),
                "expected_messages": [],
            },
            id="integrity error",
        ),
    ],
)
async def test_register_book(
    params: Mapping[str, Any],
    register_book_cmd: RegisterBook,
    uow_with_data: AbstractUnitOfWork,
):
    uow = uow_with_data
    async with uow as transaction:
        operation = await register_book(register_book_cmd, transaction)
        assert operation is not None
        res = await uow.books.by_id(register_book_cmd.id)
        if operation.is_ok():
            await transaction.commit()
        else:
            await transaction.rollback()

    assert operation == params["expected_result"]
    if res.is_ok():
        assert res.unwrap().messages == params["expected_messages"]

Run our tests

$ poetry run pytest -sxv
========================== test session starts ==========================
collected 10 items

tests/test_service_handler_add_book.py::test_register_book[ok] PASSED
tests/test_service_handler_add_book.py::test_register_book[integrity error] PASSED
tests/test_service_handler_add_book.py::test_bus_handler PASSED
tests/uow_sqla/test_repositories.py::test_book_add_ok PASSED
tests/uow_sqla/test_repositories.py::test_book_add_err[params0] PASSED
tests/uow_sqla/test_repositories.py::test_book_by_id[return a known book] PASSED
tests/uow_sqla/test_repositories.py::test_book_by_id[return an error] PASSED
tests/uow_sqla/test_repositories.py::test_eventstore_add PASSED
tests/uow_sqla/test_transaction.py::test_commit PASSED
tests/uow_sqla/test_transaction.py::test_rollback PASSED
========================== 10 passed in 0.48s ===========================

Note

When you have more than one parametrized tests, this is important to use the pytest.param function and set an id. Fixtures becomes longer than tests, and may be refactor for clarity as well. From my point of view, at the moment, it is acceptable like this.