How to parametrize fixtures

Two common challenges writting unit tests are the following: setting up an environment in case our tested code depends on external infrastructure. To test that we need to mock the dependencies and the second challenge is the generation of the test data. In this article, I will demonstrate the last one. I will show you how using the framework pytest and two of its best features: parametrizations and fixtures, we can parametrize our test functions with all shorts of test data, making our code more reusable, so we can focus on the logic that we intent to test.

Table of contents

Recommended knowledge and services we will use

• pytest
• Python

Fixtures

A test fixture (also called “test context”) is used to set up our test state and input data needed for the text execution. With pytest we can create our own fixtures. We also have a list of built-in fixtures that we can use out of the box. One that will help us later will be the fixture request. Let’s see an example of how to create a fixture to initilize our test functions.

In this example we have a function that process SQS records and translate them in a list of Order objects. We can think on a decoupled architecture, where we have a SQS queue in between systems and we want message the listeners of the queue when an order change.

"""Module to process SQS messages."""
from dataclasses import dataclass
import json
from typing import List

@dataclass
class Order:
    order_number: int
    state: str

def sqs_events_processor(events: dict) -> List[Order]:
    """Process SQS events and translating in list of orders."""
    if "Records" not in events:
        return []

    sqs_records = events["Records"]

    orders = []
    for record in sqs_records:
        if "body" not in record:
            continue

        order = json.loads(record["body"])
        orders.append(
            Order(
                order_number=order["order_number"],
                state=order["event_type"]
            )
        )

    return orders

This code can be improved and test in many ways, but it’s not the goal of this article. It’s show you how to parametrize fixtures. Next, we created a tests folder in our project to test this functionality. It will be also the module where going to create our fixture:

import pytest
from index import sqs_events_processor

@pytest.fixture(name="sqs_events")
def fixture_sqs_events():
    yield {
            "Records": [
                {
                    "messageId": "f2756a8a-ea58-4b9c-8f04-5c9253564144",
                    "body": '{"event_type": "OrderConfirmed", "order_number": 12345}',
                    "messageAttributes": {}
                },
            ]
    }

def test_processing_queue_messages(sqs_events):
    """Testing the processing of SQS messages."""

    # ACT
    orders = sqs_events_processor(sqs_events)

    # ASSERT
    assert len(orders) == 1
    assert orders[0].order_number == 12345
    assert orders[0].state == "OrderConfirmed"

With the decorator @pytest.fixture we create a fixture which simulates the payload we will have receiving events from SQS. In this case, we have just one record. Our test, receives this fixture as an argument, process them and return them as orders and we assert that the order is there with expected data. Fixtures are very powerful. If you want to dive on them you can read more about fixture here

Parametrization

Parametrization enable us to parametrize our test functions in different ways, meaning, execution the same test code using in every execution different parameters. For example, in our orders processor code, in many cases I want to assert the same things:

• The length of my list. So I know that if the SQS contain multiple orders or 1 o no orders, my function will return the expected list.
• The content of each order is correct. That’s the reason we want to have expected order number or oder state.

But effectively the test code is the same:

def test_processing_queue_messages(sqs_events):
    """Testing the processing of SQS messages."""

    # ACT
    orders = sqs_events_processor(sqs_events)

    # ASSERT
    assert len(orders) == [Number of orders]
    assert orders[0].order_number == [ExpectedOrderNumberOnTestx]
    assert orders[0].state == [ExpectedStateOnTestx]

So we want to pass multiple SQS events with different shapes and content to ensure that our function does what we expect. We will need to vary a bit the test code, but it will essentially do the same.

Let’s see one simpler example of parametrization to understand the concept before to move to the next phase:

# content of test_expectation.py
import pytest


@pytest.mark.parametrize(
  "test_input,expected",
  [
    ("3+5", 8),
    ("2+4", 6),
    ("6*9", 42)
  ]
)
def test_eval(test_input, expected):
    assert eval(test_input) == expected

If we run this test we will see the following results:

The test has been execute three times, one per each of our parametrize data.

Pro tip:

If you want to see human-readable names on your parametrize tests, you can pass use ids parameters of the parametrize decorator, like this:

# example of parametrization
@pytest.mark.parametrize(
  "test_input,expected",
  [
    ("3+5", 8),
    ("2+4", 6),
    ("6*9", 42)
  ],
  ids=["first_sum_op", "second_sum_op", "multiplication_op"]
)
def test_eval(test_input, expected):
    assert eval(test_input) == expected

so you will see the following test results:

Parametrize fixtures

Now let’s say that the data that we pass as values of our parametrizatios are fixtures. This is interesting because we can pass different types of data and reuse that test data for instance in other test cases. We can also manipulate the fixture to create new test data. In the previous example we can start extracting the test_input as fixtures:

@pytest.fixture(name="first_sum_op")
def fixture_first_sum_op():
    return "3+5"

@pytest.fixture(name="second_sum_op")
def fixture_second_sum_op():
    return "2+4"

@pytest.fixture(name="multiplication_op")
def fixture_multiplication_op():
    return "6*9"

# example of parametrization
@pytest.mark.parametrize(
  "test_input,expected",
  [
    ("first_sum_op", 8),
    ("second_sum_op", 6),
    ("multiplication_op", 42)
  ],
  ids=["first_sum_op", "second_sum_op", "multiplication_op"]
)
def test_eval(test_input, expected):
    assert eval(test_input) == expected

If we run this code, it will failed, because for the parametrization first_sum_op, second_sum_op and multiplication_op are just strings when is used on the test_eval. So, somehow we need to load the fixture from it’s name. We can achieve that using the request fixture which is a built-in fixture which contains information of the test module and we can load the fixture by name in the following way:

# example of parametrization
@pytest.mark.parametrize(
  "test_input,expected",
  [
    ("first_sum_op", 8),
    ("second_sum_op", 6),
    ("multiplication_op", 42)
  ],
  ids=["first_sum_op", "second_sum_op", "multiplication_op"]
)
def test_eval(test_input, expected, request):
    # ARRANGE
    test_input_data = request.getfixturevalue(test_input)

    # ASSERT
    assert eval(test_input_data) == expected

Now imagine another use case. Imagine that you want create a fixture that receives two parameteres and the fixture get the string representations of it’s sum. Now we want that one fixture receives multiple values. First we create a fixture. Every fixture and function of our test can receive the request fixture and access to the parameters that are passed on execution time. We use indirect parameter to tell the test, find the fixture sum_op and pass the value as parameters of the fixture in every text. With the following code we have a full parametrizable test code. In this case, instead of passing the fixture as parameter value, we are passing the fixture as function parameter with different parametrize values.

@pytest.fixture(name="sum_op")
def fixture_first_sum_op(request):
    return f"{request.param[0]}+{request.param[1]}"

@pytest.mark.parametrize(
  "sum_op, expected",
  [
    ((2, 3), 5),
    ((5, 3), 8),
    ((4, 5), 9)
  ],
  indirect=["sum_op"]
)
def test_eval_sum_operations(sum_op, expected):
    # ASSERT
    assert eval(sum_op) == expected

Let’s back to the previous example. What we want to do is the following:

1. Create a fixture that contain records. Our base SQS event with the records list.

@pytest.fixture(name="sqs_envelope")
def fixture_sqs_envelope():
    yield { "Records": [] }

2. Create different flavors of fixture: one empty records, one fixture with the initial event, one fixture with multiple order records.

@pytest.fixture(name="sqs_envelope")
def fixture_sqs_envelope():
    yield { "Records": [] }


@pytest.fixture(name="one_confirmed_order")
def fixture_one_confirmed_order(sqs_envelope):
    sqs_envelope["Records"].append(
        {
            "messageId": "f2756a8a-ea58-4b9c-8f04-5c9253564144",
            "body": '{"event_type": "OrderConfirmed", "order_number": 11111}',
            "messageAttributes": {}
        }
    )

    yield sqs_envelope

@pytest.fixture(name="two_orders")
def fixture_two_orders(one_confirmed_order):
    one_confirmed_order["Records"].append(
        {
            "messageId": "f2756a8a-ea58-4b9c-8f04-5c9253564144",
            "body": '{"event_type": "OrderCanceled", "order_number": 22222}',
            "messageAttributes": {}
        }
    )

    yield one_confirmed_order

Fixtures can recieve other fixtures and we can manipulate them to yield different values. That’s one of the powerful aspects of fixtures: the possibility of programmatically create new test data.

3. Use the fixtures in text using parametrization. For simplicity, I will only paramtrize the SQS event fixture and not the expected results

@pytest.mark.parametrize(
  "sqs_records_fixture",
  [
    "sqs_envelope",
    "one_confirmed_order",
    "two_orders"
  ]
)
def test_processing_queue_messages(sqs_records_fixture, request):
    """Testing the processing of SQS messages."""

    # ARRANGE
    sqs_records = request.getfixturevalue(sqs_records_fixture)

    # ACT
    orders = sqs_events_processor(sqs_records)

    # ASSERT
    # We can create the objects Order and pass them as parameters too.
    # Doing this for simplicity of the example.
    if sqs_records_fixture == "sqs_envelope":
        assert len(orders) == 0

    if sqs_records_fixture == "one_confirmed_order":
        assert len(orders) == 1
        assert orders[0].order_number == 11111
        assert orders[0].state == "OrderConfirmed"

    if sqs_records_fixture == "two_orders":
        assert len(orders) == 2
        assert orders[0].order_number == 11111
        assert orders[0].state == "OrderConfirmed"
        assert orders[1].order_number == 22222
        assert orders[1].state == "OrderCanceled"

You can find the code example on the code repository dedicated for the blog.