Even data engineering teams have to implement backend services with REST API. In this article, I show how to test such services in Python using a BDD implementation - Behave.
Table of Contents
- Dependencies
- Defining the test scenarios
- Implementing the scenarios
- Why do we use high-level testing?
It doesn’t matter what technology I’ve used to write the underlying service. I use it as an opaque box, so all I can do is calling the API and checking what data I receive. How do we name such testing? Some people call it end-to-end testing. Others prefer calling it integration tests or API contract tests. Let’s say it is an API contract test because I am interested in checking whether the API behaves as expected by the clients.
There are a few consequences of testing the API contract of an opaque box. First of all, I cannot mock any dependencies or directly examine the underlying database. All of my tests must use the REST API, and I must pretend I know nothing about the service implementation.
In this article, I’ll skip deployment and test environment setup. Let’s assume I’ve already deployed the application and populated a test database with the data I need.
Dependencies
To run the tests, we need a few dependencies. I use poetry as the dependency manager, so my dependencies look like this:
[tool.poetry.dev-dependencies]
behave = "^1.2.6"
pytest = "^6.2.2"
Defining the test scenarios
The application we test is a Twitter clone. First, I have to write test scenarios using Gherkin. It is not an article about writing Gherkin scenarios, so I implement only two scenarios:
Feature: User feed contains tweets posted by followed users
#Rule: User's feed contains tweets posted by people followed by the user
Scenario: User who doesn't follow anyone doesn't see tweets
Given Alice doesn't follow anyone
When Alice retrieves the feed
Then Alice sees an empty list
Scenario: User sees tweets posted by followed accounts
Given Alice follows Bob
And Bob posted a tweet
When Alice retrieves the feed
Then Alice sees the content posted by Bob
When we use Behave, we store the scenarios in a .feature file in the features directory.
Implementing the scenarios
Now, I implement the BDD steps. We must store the step implementations in the feature/steps directory.
In the beginning, I have to produce random tweets. I prefer to use random values here because leftovers from the previous tests won’t spoil the results even if I don’t clean the test environment between tests. Still, it is better to redeploy everything and purge the test database.
import random
import string
from behave import *
from features.steps.api_client import *
def _random_string(n):
return ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(n))
For the first scenario, we need three functions:
@given("Alice doesn\'t follow anyone")
def alice_unfollows_all(context):
unfollow_all('Alice')
@when("Alice retrieves the feed")
def alice_retrieves_feed(context):
context.feed = retrieve_tweets('Alice')
@then("Alice sees an empty list")
def feed_is_empty(context):
print(context.feed)
assert len(context.feed) == 0
We print the feed content in the then implementation because Behave captures the standard output and prints it when a test fails.
We need to implement the two helper functions, unfollow_all and retrieve_tweets. The actual implementation doesn’t matter, so I do not post the implementations of other helper functions.
import requests
def _make_url(URL):
pass # here we should return the URL to the test environment
def _get_pwd(user):
pass # here we return the test password of the test account
def unfollow_all(username):
requests.delete(_make_url('following'), auth=(username, _get_pwd(username)))
def retrieve_tweets(username):
response = requests.get(_make_url('tweets'), auth=(username, _get_pwd(username))
return response.json()
In the second scenario, we have to post a tweet and see whether we get it while retrieving the feed:
@given("Alice follows Bob")
def alice_follows_bob(context):
follow('Alice', 'Bob')
@given('Bob posted a tweet')
def bob_posts_tweet(context):
context.expected_tweet = _random_string(20)
post_tweet('Bob', context.expected_tweet)
# we reuse the alice_retrieves_feed function
@then('Alice sees the content posted by Bob')
def alice_sees_bob_post(context):
filtered_feed = [x for x in context.feed if x['username'] == 'Bob' and x['tweet'] == context.expected_tweet]
assert len(filtered_feed) == 1
Why do we use high-level testing?
What is the benefit of testing service through its API without mocking any dependencies? At first, it seems to be a useless testing method because when a test fails, we don’t know the reason for the failure. Of course, API contract tests are not enough. In addition to them, we need lots of unit tests to verify the internals of the application.
The API contract tests, however, give us other benefits:
- we avoid an embarrassing situation when all unit tests pass, but the application doesn’t work which happens when we test individual components, but we fail to verify interactions between components
- we’ll not break the client application. If we make a modification and unintentionally change the API, we have a chance to detect it using the API contract tests before we deploy it and break the contract.
- we can provide multiple implementations of the same API that are interchangeable and compatible with each other - it is useful when we write a new version of a service or split a large application into smaller services