January 10th, 2022

Testing a Flask API





This is part of a series of articles on SaintsXCTF Version 2.0. The first article in the series provides an overview of the application. You DO NOT need to read prior articles in the series to fully understand this article.

In a prior article, I discussed my Flask API written for SaintsXCTF, an application which allows users and their teammates to log running exercises and keep track of their running mileage. As with any piece of software, this Flask API requires extensive testing to keep it functional and maintainable. In this article, I discuss the automated testing approach I implemented for the API.

The tests for my application use the built-in Python unittest library, along with the coverage library for code coverage. Code coverage determines which lines of API source code are covered by tests, allowing me to find gaps in testing. The coverage library shows both textual and HTML reports of code coverage, making it extremely easy to drill down into specific source files and find which lines of code aren't being tested.

In the Flask application source code, I created a custom CLI command which triggers the unittest tests with code coverage enabled. This way, application tests are triggered from the command line with a flask test command. In Flask code, the CLI command is a Python function test(), located within a file. test() is added to the Flask application in my file with the following line:


test() starts code coverage, runs the tests, and generates a code coverage report. The code is shown below, and is inspired by a function found in the book Flask Web Development1.

cov = None if os.environ.get('FLASK_COVERAGE'): cov = coverage.coverage(branch=True, include=['', 'dao/*', 'model/*', 'route/*', 'utils/*']) cov.start() @click.command() @with_appcontext def test(): """ Create a Flask command for running unit tests. Execute with 'flask test' from a command line. """ if not os.environ.get('FLASK_COVERAGE'): os.environ['FLASK_COVERAGE'] = '1' os.execvp(sys.executable, [sys.executable] + sys.argv) # Create a test runner an execute the test suite tests = unittest.TestLoader().discover('tests') runner = unittest.TextTestRunner(verbosity=3) result: unittest.TestResult = if cov: cov.stop() print('Coverage Summary:') basedir = os.path.abspath(os.path.dirname(__file__)) cov_dir = os.path.join(basedir, 'tmp/coverage') cov.html_report(directory=cov_dir) print('HTML version: file://%s/index.html' % cov_dir) cov.erase() exit(len(result.errors))

test() runs all the Python files in the test directory of my application.

Most of the tests in my application invoke API endpoints and assert they return certain HTTP response bodies and response status codes. This means that in order for my tests to execute successfully, a running instance of the API is needed. Instead of using the API running in production, I use a test API that is quickly spun up before tests begin and spun down after tests stop. All the configuration for this test API and the execution environment for the tests themselves is accomplished using Docker and Docker Compose.

In general, each component of my application backend has a corresponding Docker Compose file. There is also a Docker Compose file for the API tests. These Docker Compose files exist in a docker-compose directory in my repository.

There are five Docker Compose files in total, each running a single Docker container. There are four Docker Compose files that together complete my application backend. They are for my database, authentication API, AWS Lambda function API, and main Flask API. These four Docker Compose configurations need to be started using the docker-compose up command before the API tests are run. Once they are running, the API test Docker Compose file can be run. This Docker Compose file starts a container, runs the API tests on that container, and then stops the container. Its configuration is shown below.

# docker-compose-test-local.yml version: '3.7' services: test: build: context: ../../api/src/ dockerfile: local.test.dockerfile network: host networks: - local-saints-xctf-api-net networks: local-saints-xctf-api-net: driver: bridge

This Docker Compose file runs a local.test.dockerfile Dockerfile, which has the following configuration:

FROM python:3.8-alpine RUN apk update \ && apk add --virtual .build-deps gcc python3-dev libc-dev libffi-dev g++ \ && pip install --upgrade pip \ && pip install pipenv COPY . /src WORKDIR /src RUN pipenv install ENV FLASK_APP ENV ENV localtest COPY credentials .aws/ ENV AWS_DEFAULT_REGION us-east-1 ENV AWS_SHARED_CREDENTIALS_FILE .aws/credentials ENTRYPOINT ["pipenv", "run", "flask", "test"]

At its core, this container uses a python:3.8-alpine container as its base, installs necessary dependencies for the API code, and runs the flask test command in the container ENTRYPOINT. Running this container shows a console output similar to the following:

Starting docker-compose_test_1 ... done Attaching to docker-compose_test_1 test_1 | test_app_exists (test_src.testApp.TestApp) ... ok test_1 | test_non_existent_route (test_src.testApp.TestApp) test_1 | ... test_1 | test_user_update_last_login_by_username_put_route_unauthorized (test_src.test_route.testUserRoute.TestUserRoute) test_1 | Test performing an unauthorized HTTP PUT request on the '/v2/users/<username>/update_last_login' route. ... ok test_1 | test_1 | ---------------------------------------------------------------------- test_1 | Ran 467 tests in 77.992s test_1 | test_1 | OK (skipped=11) test_1 | Coverage Summary: test_1 | Name Stmts Miss Branch BrPart Cover test_1 | ---------------------------------------------------------------- test_1 | 73 35 6 0 48% test_1 | dao/ 25 0 0 0 100% test_1 | dao/ 15 0 0 0 100% test_1 | dao/ 40 3 0 0 92% test_1 | model/ 75 0 0 0 100% test_1 | model/ 30 2 4 2 88% test_1 | route/ 486 31 138 30 90% test_1 | utils/ 19 2 14 2 88% test_1 | ... test_1 | ---------------------------------------------------------------- test_1 | TOTAL 3625 261 692 154 90% test_1 | HTML version: file:///src/tmp/coverage/index.html docker-compose_test_1 exited with code 0

This output states that the container started, 467 tests ran in 78 seconds with a successful (OK) status, and the code coverage of the API is 90%.

The test code for my API exists within a tests directory. The file structure of the tests directory matches the application source code directory. For example, inside tests is a test_src directory, which matches the src directory of the application. Inside test_src is a file, which is a test file corresponding to an file in the source code. As you can see, in the test code, directories and files are prefixed with test. Directories are in snake case while files are in camel case. This isn’t a required convention, but consistent naming makes it easy to associate test files with source code.

The contents of, which tests the entrypoint of the Flask application, is shown below.

from flask import current_app, Response from tests.TestSuite import TestSuite class TestApp(TestSuite): def test_app_exists(self): self.assertTrue(current_app is not None) def test_non_existent_route(self) -> None: """ Test performing an HTTP GET request against an endpoint that doesn't exist. This query should invoke the custom 404 error handler. """ response: Response = self.client.get('/path/doesnt/exist') response_json: dict = response.get_json() self.assertEqual(response.status_code, 404) self.assertEqual(response_json.get('error_description'), 'Page Not Found') self.assertGreater(len(response_json.get('exception')), 0) self.assertEqual(response_json.get('contact'), '') self.assertEqual(response_json.get('api_index'), '/versions')

This test code contains two test functions: test_app_exists() and test_non_existent_route(). test_app_exists() checks if current_app, a proxy to the Flask application object, exists as expected2. test_non_existent_route() tests that making an HTTP request to an invalid route in the API returns a specific JSON response with a 404 status code.

Each test file contains a single class with functions for individual tests. These test classes, TestApp in the example above, extend a base class called TestSuite. TestSuite performs setup and cleanup work for tests. TestSuite extends a unittest.TestCase class, which exposes methods for performing setup and cleanup logic3. TestSuite overrides these methods, as shown below. The full code is available in my file.

import unittest import os import asyncio import json import boto3 import aiohttp from flask.testing import FlaskClient from config import config from app import create_app from database import db class TestSuite(unittest.TestCase): jwt = None auth_url = None @classmethod def setUpClass(cls) -> None: """ Set up logic performed before each test class executes. """ flask_env = os.getenv('FLASK_ENV') or 'local' auth_url = config[flask_env].AUTH_URL TestSuite.auth_url = auth_url secretsmanager = boto3.client('secretsmanager') response = secretsmanager.get_secret_value(SecretId=f'saints-xctf-andy-password') secret_string = response.get("SecretString") secret_dict = json.loads(secret_string) client_secret = secret_dict.get("password") async def token(): async with aiohttp.ClientSession() as session: async with url=f"{auth_url}/token", json={'clientId': 'andy', 'clientSecret': client_secret} ) as response: response_body = await response.json() result = response_body.get('result') if result: TestSuite.jwt = result def setUp(self) -> None: """ Set up logic performed before every test. """ if os.environ.get('ENV') == 'localtest': env = 'localtest' else: env = 'test' = create_app(env) self.app_context = self.app_context.push() self.client: FlaskClient = self.jwt = TestSuite.jwt self.auth_url = TestSuite.auth_url self.jwts = {} def tearDown(self) -> None: """ Tear down logic performed after every test. """ db.session.remove() self.app_context.pop()

The setup functions perform actions such as retrieving an authentication token (JWT) to use in tests and creating a test client for the API of type FlaskClient. More expensive operations, such as making an external API call to receive a JWT, are performed in setUpClass() instead of setUp(). This way, long running tasks don’t take place before each test function, only at the start of each class.

I already showed some basic tests in the file, but now let's look at more complex examples. is a file that tests all the API routes defined in my source code file. Let’s walk through a few tests, starting with test_group_get_all_route_200().

def test_group_get_all_route_200(self) -> None: """ Test performing an HTTP GET request on the '/v2/groups/' route. This test proves that the endpoint returns a list of groups. """ response: Response = self.client.get('/v2/groups/', headers={'Authorization': f'Bearer {self.jwt}'}) response_json: dict = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(response_json.get('self'), '/v2/groups') self.assertGreater(len(response_json.get('groups')), 1)

This test checks that API calls to the /v2/groups/ endpoint returns a 200 status code and a specific JSON body. It is very similar to the test_non_existent_route() test, except this API endpoint requires authentication, which is passed to the endpoint in the Authorization header.

A slightly more complex test is test_group_by_id_put_route_400_not_an_admin().

def test_group_by_id_put_route_400_not_an_admin(self) -> None: """ Test performing an HTTP PUT request on the '/v2/groups/<group_id>' route. This test proves that a user trying to update a group they are not an administrator of results in a 400 error. """ response: Response = self.client.get( '/v2/groups/saintsxctf/wmenstf', headers={'Authorization': f'Bearer {self.jwt}'} ) response_json: dict = response.get_json() self.assertEqual(response.status_code, 200) self.assertIsNotNone(response_json.get('group')) group_id = response_json.get('group').get('id') response: Response = self.client.put(f'/v2/groups/{group_id}', headers={'Authorization': f'Bearer {self.jwt}'}) response_json: dict = response.get_json() self.assertEqual(response.status_code, 400) self.assertEqual(response_json.get('self'), f'/v2/groups/{group_id}') self.assertFalse(response_json.get('updated')) self.assertIsNone(response_json.get('group')) self.assertEqual( response_json.get('error'), f'User andy is not authorized to update a group with id {group_id}.' )

This test asserts that if my user tries to update a group it is not a member of (the Women’s Track & Field group), the API returns a 400 error. Two API calls are needed to complete the test, which adds some complexity. The first API call (a GET request) receives Women’s Track & Field group information, including a unique identifier for the group. This unique identifier is stored in group_id and is used in the next API call, a PUT request to /v2/groups/{group_id}.

A successful variant of the test above is test_group_by_id_put_route_200(), which asserts that my user can update an Alumni group, which I am a member of. This test also requires two API calls.

def test_group_by_id_put_route_200(self) -> None: """ Test performing an HTTP PUT request on the '/v2/groups/<group_id>' route. This test proves that if the updated group object is valid and the user is an admin of the group, the group is updated and a 200 status code is returned. """ response: Response = self.client.get( '/v2/groups/saintsxctf/alumni', headers={'Authorization': f'Bearer {self.jwt}'} ) response_json: dict = response.get_json() self.assertEqual(response.status_code, 200) self.assertIsNotNone(response_json.get('group')) group_id = response_json.get('group').get('id') group_dict: dict = response_json.get('group') group_dict['description'] = f"Updated: {}" request_body = json.dumps(group_dict) response: Response = self.client.put( f'/v2/groups/{group_id}', data=request_body, content_type='application/json', headers={'Authorization': f'Bearer {self.jwt}'} ) response_json: dict = response.get_json() self.assertEqual(response.status_code, 200) self.assertEqual(response_json.get('self'), f'/v2/groups/{group_id}') self.assertTrue(response_json.get('updated')) self.assertIsNotNone(response_json.get('group'))

All the API endpoint tests in my application follow a similar pattern to the ones I’ve shown. There are two other types of tests for my Flask application. The first type is unit tests for utility functions in my API, such as those found in This file tests conversions between distance units, such as converting miles to kilometers. The second type is unit tests for SQLAlchemy Model classes. SQLAlchemy model class tests are located in a test_model directory. One example is, which tests the model.

from datetime import datetime from tests.TestSuite import TestSuite from model.User import User class TestUser(TestSuite): user1_dict = { 'username': "andy", 'first': 'Andy', 'last': 'Jarombek', 'salt': None, 'password': 'hashed_and_salted_password', 'profilepic': None, 'profilepic_name': None, 'description': "Andy's Profile", 'member_since': datetime.fromisoformat('2016-12-23'), 'class_year': 2017, 'location': 'Riverside, CT', 'favorite_event': '8K, 5000m', 'activation_code': 'ABC123', 'email': '', 'subscribed': 1, 'last_signin': datetime.fromisoformat('2019-12-10'), 'week_start': 'monday', 'deleted': False } user2_dict = { 'username': "andy2", 'first': 'Andrew', 'last': 'Jarombek', 'salt': None, 'password': 'hashed_and_salted_password', 'profilepic': None, 'profilepic_name': None, 'description': None, 'member_since': datetime.fromisoformat('2019-12-10'), 'class_year': 2017, 'location': None, 'favorite_event': None, 'activation_code': 'DEF456', 'email': '', 'subscribed': None, 'last_signin':, 'week_start': None, 'deleted': False } user1 = User(user1_dict) user1copy = User(user1_dict) user2 = User(user2_dict) def test_user_str(self) -> None: """ Prove that the human readable string representation of a User object is as expected. """ log_str = 'User: [username: andy, first: Andy, last: Jarombek, salt: None, ' \ 'password: hashed_and_salted_password, ' \ "description: Andy's Profile, member_since: 2016-12-23 00:00:00, class_year: 2017, " \ 'location: Riverside, CT, favorite_event: 8K, 5000m, activation_code: ABC123, ' \ 'email:, subscribed: 1, last_signin: 2019-12-10 00:00:00, week_start: monday, ' \ 'deleted: False]' self.maxDiff = None self.assertEquals(str(self.user1), log_str) self.assertEquals(self.user1.__str__(), log_str) def test_user_repr(self) -> None: """ Prove that the machine readable string representation of a User object is as expected. """ self.assertEquals(repr(self.user1), "<User 'andy'>") self.assertEquals(self.user1.__repr__(), "<User 'andy'>") def test_user_eq(self) -> None: """ Prove that two User objects with the same property values test positive for value equality. """ self.assertTrue(self.user1 == self.user1copy) self.assertTrue(self.user1.__eq__(self.user1copy)) def test_user_ne(self) -> None: """ Prove that two User objects with different property values test negative for value equality. """ self.assertTrue(self.user1 != self.user2) self.assertTrue(self.user1.__ne__(self.user2))

There are a couple aspects of the model classes that I prioritize in these tests. test_user_str() tests that the string representation of a model class (created by passing an instance of the class to str()) is as expected. test_user_repr() tests that the machine representation of a model class (created by passing an instance of the class to repr()) is as expected. test_user_eq() and test_user_ne() test for value equality between instances of the model class.

One piece of my API that is not directly under test is the Data Access Objects (DAOs). My reasoning for this decision is that the DAOs are invoked by the API routes, so they are already indirectly tested by the API endpoint tests. Testing them individually would be a duplication of these existing endpoint tests, just written in a different format.

Writing automated tests is critical for the reliability of any application, and hopefully this article helped illustrate that tests for Flask APIs are easy to set up and write. All the code discussed in this article is available in my saints-xctf-api repository on GitHub.