In this final part of our series: sharing our example JSON payloads across our documentation and tests, and solid ways of testing our endpoints!

Managing Payload Examples

Now let’s consider JSON payloads we use for tests and documentation. It’d be good practice to make sure that the tests and docs stay in sync and that we test the things we document. We can label our payloads with descriptive names and retrieve them when we need to. Note that for the purposes of our documentation we need this JSON to be available to the application, so the easiest thing to do, deployment-wise, is to put it in the resources directory.

We could either store a separate JSON file for each example, named by our test-case description, or we could put them all in a dictionary whose key is the description. At first, putting them all in a single JSON dictionary seems like the easiest way to go. However, such a JSON file would obviously need to be parseable. This means that any error test cases we want to create that have invalid JSON would need to use a different mechanism. Let’s therefore put the test payloads each in its own file under resources/examples.

You can read about the various ways you can include example data in the docs here. The richest method is here. I say it’s the richest because it includes not only the payload itself, but summary and description fields. The summary and description will show up in our Swagger and ReDoc GUIs. Woo!

ExamplePayloads

Similar to my approach to MarkdownSlicer, I’ve created a simple utility class called ExamplePayloads to help us reuse our JSON for both the docs and the tests. These examples live as .json files in resources/examples in our repo. ExamplePayloads uses the OpenAPI metadata format you can see if you follow that last link. We can easily grab a test payload by name for our pytest code, and can also specify a list of payloads by name for our REST API docs. 🤓

Here’s an example of using the example payloads in a test case:

async def test_error_user_create_type_error(self,
http_client: AsyncClient,
examples: ExamplePayloads
) -> None:
“””
Ensure that an error is returned if a wrong company type is provided
when creating a user.
“””
payload = examples.get_example_value(‘error-user-create-john-doe-company-type-error’)
response = await http_client.post(“/v1/users”, json=payload)

assert response.status_code == 422
detail: dict = response.json()[“detail”] assert detail[0][“type”] == “int_parsing”
assert detail[0][“msg”] ==
“Input should be a valid integer, unable to parse string as an integer”
assert detail[0][“input”] == “garbage”
assert detail[0][“loc”] == [“body”, “company_id”]

For our endpoint documentation we simply specify the list of relevant examples when we define our endpoint function, like this:

@rest.post(“/v1/users”,
response_model=UserRead,
tags=[“Users”],
status_code=status.HTTP_201_CREATED,
description=md_slicer.get_endpoint_docs(“POST /v1/users/”))
async def create_user(*,
session: Session = Depends(session_generator),
user: Annotated[
UserCreate,
Body(openapi_examples=examples.get_examples([
“user-create-john-doe”,
“error-user-create-john-doe-company-type-error”,
“error-user-create-john-doe-no-password”,
“error-user-create-john-doe-unknown-company”]))
]) -> User:

In Swagger, it will look like this:

while in ReDoc you’ll see:

Testing

Writing good test cases is one of the most critical parts of creating solid code. My tests in the GitHub repo for this project aren’t as comprehensive as the ones for the REST API I wrote for my last startup, but they’re definitely a good start. There are some tricks and subtleties that I make use of that I’ll cover in this section.

I always write tests as I go, and like to test the stack of code from the bottom up to make it easier to write and debug the system. We’ll start with some of the lower-level parts of the code.

EndpointTestFixtures

The EndpointTestFixtures class contains various Pytest fixtures. These are functions that Pytest calls automatically when it runs a test case, to create resources that the test can use. These resources simply have to be added to a test function’s parameter list in order to be created and passed in.

http_client()

This fixture creates an httpx.AsyncClient for testing our endpoints. Ours has the feature that it can either directly call into our FastAPI dispatcher, without requiring us to launch an instance of our server, or if you set the TEST_SERVER_HOST environment variable it can run the tests against a running instance such as a staging server in your CI/CD environment. This lets us reuse our functional tests as smoke tests for our server deployments. 🥳

examples()

The examples() fixture loads the set of example JSON payloads that we use for both our tests and our docs, to make it easy to use them in our test cases.

set_up_db()

set_up_db() uses SQLModel’s create_all() function to create all the tables for our application entities, if they don’t exist. It’s declared with:

@pytest.fixture(scope=”module”, autouse=True)

so that it runs once per test module. As you’ll see, our db tests are written so that they run in isolated transactions and otherwise clean up after themselves. We do this so that they can be run in parallel and can be run against a database that’s already populated.

ephemeral_session()

This is a key part of our database tests. This fixture creates a nested, ephemeral database session that we use in our db tests. It automatically encapsulates the db actions that we take (such as adding or deleting rows) into a transaction, and then automatically rolls them back when the test exits (either normally, or by raising an Exception).

There are some key things to note here:

To speed up our local testing while also doing a quick check for race conditions and other conflicts in our code we run tests in parallel. We enable this via:pip install pytest-xdist==3.5.0Using the -n auto command-line parameters to pytest. In PyCharm, you go to your pytest Run Configuration (e.g., by right-clicking on the tests/ dir and choosing More Run/Debug –> Modify Run Configuration) and putting it in Additional Arguments.The with context manager ephemeral_transaction_scope(), which handles the rollback(). Neither it nor our tests should call commit(). If they do, they will commit any changes to the database. This can cause conflicts between our test cases, since they are written to assume that they are acting on a clean database, at least as far as the test data is concerned. As an example, if two tests commit() a User with email foo@bar.com, one of them will fail due to the UniqueConstraint on the email field.

tests/entities/test_search_entities.py

Let’s take a look now at the tests for the SearchModel classes.

class TestPydanticErrorChecking

TestPydanticErrorChecking tests and demonstrates that we’ll get ValidationError exceptions if the payload the client sends is bad. It tests for values of the wrong type, unknown fields, and other errors. It also shows that these are detected both at object instantiation time, and at assignment time since we specified validate_assignment = True in our base class Config:

class Config:
extra = “forbid” # raise a ValidationError if we see unknown fields
validate_assignment = True # validate the model at runtime when the model is changed

class TestSearchEntitySerializationDeserialization

TestSearchEntitySerializationDeserialization contains tests for the basic serialization / deserialization functionality for single objects and trees of objects. These tests specify the exact SearchModel subclass that they are deserializing into, so they are not testing polymorphic deserialization.

class TestSearchModelPolymorphicDeserializationAndSearch

Recall from earlier that we implemented polymorphic deserialization so that we can write endpoints that have SearchModel as our payload type. This class contains tests that verify that payloads are deserialized into the correct class (that is, that our discriminator() validator function is having the right effect).

It also tests all of our example database searches. Let’s take a look at one of these, test_or_search_model(), which tests the Or condition:

from tests.helpers.test_entities_helpers import add_test_entities_john_doe
from tests.helpers.test_entities_helpers import (validate_john_doe_1,
validate_john_doe_1000000042, validate_jane_doe_1000000024, …

def test_or_search_model(self,
ephemeral_session: Session,
examples: ExamplePayloads
) -> None:
payload = examples.get_example_value(“user-search-by-name-or-company-id”)
model = parse_obj_as(SearchModel, payload)
assert isinstance(model, OrSearchModel)
assert model.dict(exclude_unset=True) == payload

assert len(model.children) == 2
assert model.children[0].last_name == “Doe”
assert model.children[0].company_id is None

assert model.children[1].first_name is None
assert model.children[1].company_id == 1000000024

# Ok, the tree looks good. Let’s test the search in the db!

condition = model.render_condition()

add_test_entities_john_doe(ephemeral_session)

results = ephemeral_session.query(User).filter(condition).all()
assert len(results) == 3

user_0 = results[0] validate_john_doe_1000000042(user_0)

user_1 = results[1] validate_john_doe_1(user_1)

user_2 = results[2] validate_jane_doe_1000000024(user_2)

Here’s the sequence of events:

We import tests.common.EndpointFixtures, which in turn imports rest_api.db.common which in turn sees that we don’t have the DATABASE_URL environment variable set, so it creates a one-time SQLite database for us in /tmp.Pytest runs our module-level fixture create_tables_if_missing(), which creates our entity tables.Pytest runs our fixtures ephemeral_session() and examples() and calls our test function, passing the fixture results in as parameters. ephemeral_session() creates a nested database session for us, and will manage rollback when our test case exits.We get the test JSON payload from our examples object.We call Pydantic’s parse_obj_as() with SearchModel as the expected type.
• SearchModel is a Union of all of our search model types. Pydantic tries each one of the Union’s subtypes one by one, until it finds one that validates successfully. Our discriminator() validator function checks whether or not the type of the object being validated matches the value of the type field being deserialized.We check the SearchModel tree that was deserialized to make sure it looks right.We call model.render_condition() to get the SQLAlchemy representation of the where clause for our SearchModel expression (SQLAlchemy’s AST).We call a utility function, add_test_entities_john_doe(), passing it the ephemeral_session database session that our fixture set up. This adds the test Users we need for our test. Note that we don’t commit(), so these INSERTs will be rolled back when we’re done.We run the query in the database.We validate that the correct results were returned.We return from our test, and our database changes are rolled back for us automagically because we did them in ephemeral_session.

tests/entities/test_select_generation.py

This file deserializes a bunch of test search payloads, calls our render_condition() function to get the SQLAlchemy condition AST, calls SQLAlchemy to get the actual SQL statement, and validates that it’s correct.

tests/helpers/test_api_docs_helpers.py

Tests for our MarkdownSplicer class.

tests/helpers/entities_helpers.py

Utility functions for our tests to:

add test data to our dbvalidate objects created from that test data

tests/api/test_api_users.py

This is where the rubber hits the road: it contains tests for all of our /v1/users endpoints.

Remember that our EndpointTestFixtures class has two critical test fixtures for endpoint testing:

http_client() gives us a client that can either directly call FastAPI’s request dispatch mechanism, so we don’t need a running server, or if we set the TEST_SERVER_HOST environment variable to the URL for a host (e.g., http://localhost:8000) it can run our tests against a running server.examples() gives us the JSON payloads we use for both tests and docs.

An important thing to note is that when we test the endpoints with a running server, they will all write to the same database. Even when they clean up after themselves, the operation as a whole is not in a transaction. This causes race conditions, as we’ll see below.

Our tests should clean up after themselves (e.g., if we CREATE a Company, we should DELETE it at the end). If you look at test_user_crud(), it does most of the functionality in a try block, and has the delete() call in a finally block. This ensures that if something goes wrong and the main part of the test fails, it’ll still delete the test User. In this way, we can run tests against a server with a populated database (e.g., our staging server) safely, knowing that our tests haven’t left garbage around that can conflict when they are re-run.

class TestExamples

Tests our examples payloads, including error reporting for bad payloads.

Testing Against a Local Server

As we saw above, our EndpointFixtures Pytest class can create an http_client that can test against a server. Let’s think about that case.

Our ephemeral_session magic lets us run our tests in parallel against a local sqlite database, because each test does its database INSERTs inside a transaction that gets rolled back for us automatically, and our transactions are isolated because we specified the highest isolation level when we created our database engine.

However, this magic doesn’t happen in a running server, since the database operations aren’t in a single transaction that gets rolled back. Shen we do in INSERT in a test it will get committed to the database by the server. Any other concurrently-running test will see the inserted data, until the test cleans up after itself. Therefore, our tests in tests/api, which hit the endpoints, cannot be run concurrently. In other words, when we’re testing against any live server we can’t use -n auto to run those tests to speed up the testing time.

We could write the tests so that they can’t conflict with each other. This is tricky, though, because we check things like the number of search results that are returned for our searches. If we concurrently insert rpeck-0@rpeck.com and rpeck-1@rpeck.com in two different tests, and a concurrently-running test searches for all the users @rpeck.com and counts them, the answer it gets will vary because of the race condition.

I think the restriction that we can’t run the tests in parallel against a running server is better than complexifying the tests. If necessary, we can always run the tests other than the API tests in parallel, and only serialize the ones in test/api.

Wrapping Up!

I hope this material and the repo have been helpful! I know that it’ll help me remember all the obscure tricks when I need them. 🤓

If you’ve got any questions or feedback, feel free to comment or email me at rpeck@rpeck.com.

Best Practices for Modern REST APIs in Python, Part 4 of 4 was originally published in Level Up Coding on Medium, where people are continuing the conversation by highlighting and responding to this story.

​ Level Up Coding – Medium

about Infinite Loop Digital

We support businesses by identifying requirements and helping clients integrate AI seamlessly into their operations.

Gartner
Gartner Digital Workplace Summit Generative Al

GenAI sessions:

  • 4 Use Cases for Generative AI and ChatGPT in the Digital Workplace
  • How the Power of Generative AI Will Transform Knowledge Management
  • The Perils and Promises of Microsoft 365 Copilot
  • How to Be the Generative AI Champion Your CIO and Organization Need
  • How to Shift Organizational Culture Today to Embrace Generative AI Tomorrow
  • Mitigate the Risks of Generative AI by Enhancing Your Information Governance
  • Cultivate Essential Skills for Collaborating With Artificial Intelligence
  • Ask the Expert: Microsoft 365 Copilot
  • Generative AI Across Digital Workplace Markets
10 – 11 June 2024

London, U.K.