In this article, we will look at simple, straightforward, and practical ways of automated testing of the flask REST API we built in this previous article. The completed project can be found in the following github repository: https://github.com/tmsdev82/collectors-items-at-rest
Why continuous testing
The main motivation for having continuous testing is that it instantly lets you know when a code change has broken some (piece of) functionality. It also lowers the bar to do major code refactoring, because we are assured that the tests are continuously running and if something goes wrong we immediately know about it before it becomes a problem. It will also motivate us to write code that is unit testable, which, in turn, encourages cleaner code.
What we will learn
We will learn how to set up continuous testing and reporting of code coverage. To do all that, we will use the following python libraries:
- pytest: is a test runner which will collect tests from the project and run them
- pytest-cov: produces a test code coverage report
- pytest-watch: is a tool for re-running tests with pytest everytime a file is changed and saved
- flask-testing: helps facilitate unit testing for flask
We will be testing the following aspects of the backend API solution:
- Database model
- REST API endpoints
Package installation
Start by installing the packages mentioned in the previous section:
pip install pytest pytest-cov pytest-watch flask-testing
Don’t forget to add these dependencies to the requirements.txt file:
pip freeze > requirements.txt
Preview in new tab(opens in a new tab)
Tests
Setting up the folder structure
First add a directory to place the tests in for our automated testing flask API project. We’re going to create a similar structure to the app folder under the root and name it ‘tests’ as shown below::
collectors_items-at-rest
+--app
+--main
+--collectors_item
+--controllers
+--services
+--models
+--schemas
+--tests
+ base.py
+--collectors_item
+--controllers
+ test_collectors_item_controllers.py
+--models
+ test_collectors_item_models.py
+--schemas
+ test_collectors_item_schemas.py
The pytest tool will find any file prefixed with test_ or suffixed with _test.py.
Note: To avoid issues when writing and running tests where files in the tests folder can’t find the app module, create a setup.py file and install the modules in the virtual environment we’re using:
collectors_items-at-rest +--app +--tests +--venv + setup.py
Contents of setup.py:
from setuptools import setup, find_packages setup(name="app", packages=find_packages())
Then simply run the following install command in your project root directory:
pip install -e .
This will find all the modules and install them. This allows modules to find each other for import purposes.
Base test case
Earlier, we added a base.py file in the tests directory. This file contains the BaseTestCase class other test classes will derive from. Copy the below code snippet into base.py.
from flask_testing import TestCase
from app.main import _db
from app import blueprint
from app.main import create_app
class BaseTestCase(TestCase):
def create_app(self):
app = create_app("test")
app.register_blueprint(blueprint)
app.app_context().push()
return app
def setUp(self):
_db.create_all()
def tearDown(self):
_db.session.remove()
_db.drop_all()
Notice that what we’re importing are objects and functions from app and app.main, this is so that we can set up a flask environment for each test to work with.
The BaseTestCase class derives from TestCase from the flask_testing library so that for each test, the create_app function, which will set up the flask app, gets called. Otherwise, we would not be able to use the _db object and endpoint definitions and the like, which all require a running flask application.
In create_app, we create an app instance using the “test” configuration and then call functions to enable the endpoints we’ve defined with blueprint.
The setUp function is called before each test executes and will create a new database for each test. The tearDown function is called after each test; it closes any lingering sessions and drops the database. These two functions combined will ensure that each test starts with a new, clean database.
With the BaseTestCase completed, we are ready to write the actual tests for automated testing of our flask REST API.
Test models
tests/collectors_items/models/test_collectors_item_models.py
The first tests we’ll write concern the data model and the database. We’ll test the creation of a collectors_item record in the database and then the serialization of a database object.
from datetime import datetime
from app.main import _db
from app.main.collectors_item.models import collectors_item_model
from tests.base import BaseTestCase
class TestCollectorsItemModel(BaseTestCase):
def test_create_collectors_item(self):
collectors_item_dbo = collectors_item_model.CollectorsItemModel(
name="test_collectors_item",
description="collectors_item for testing",
collectors_item_type="ccc2",
date_added=datetime.utcnow(),
)
_db.session.add(collectors_item_dbo)
_db.session.commit()
self.assertFalse(collectors_item_dbo.id is None)
We import the database instance object _db here to be able to do database operations. We also need the collectors_item_model module to access datatable classes, and the BaseTestCase to derive the test class from.
Creating a test class like TestCollectorsItemModel can be regarded as a grouping of tests under a certain subject. Each function in the class is an individual test.
In the test_create_collectors_item(), we’ll simply create an instance of the collectors_item database object collectors_item_dbo and use the _db object to save it to the database. The collectors_item_dbo object should be updated after its data is saved to the database, so we can check if the id attribute has been updated and is no longer None.
Next, we’ll add the serialization test:
...
def test_serialize_collectors_item(self):
date_added = datetime.utcnow()
collectors_item_dbo = collectors_item_model.CollectorsItemModel(
name="test_collectors_item",
description="collectors_item for testing",
collectors_item_type="ccc2",
date_added=date_added,
)
_db.session.add(collectors_item_dbo)
_db.session.commit()
serialized_collectors_item = collectors_item_dbo.serialize()
self.assertEqual(serialized_collectors_item["id"], collectors_item_dbo.id)
self.assertEqual(serialized_collectors_item["name"], collectors_item_dbo.name)
self.assertEqual(serialized_collectors_item["description"], collectors_item_dbo.description)
self.assertEqual(serialized_collectors_item["collectors_item_type"], collectors_item_dbo.collectors_item_type)
self.assertEqual(serialized_collectors_item["date_added"], str(date_added))
# test for nullable field
collectors_item_dbo = collectors_item_model.CollectorsItemModel(
name="test_collectors_item2",
collectors_item_type="ccc2",
date_added=date_added,
)
_db.session.add(collectors_item_dbo)
_db.session.commit()
serialized_collectors_item = collectors_item_dbo.serialize()
self.assertEqual(serialized_collectors_item["id"], collectors_item_dbo.id)
self.assertEqual(serialized_collectors_item["name"], collectors_item_dbo.name)
self.assertEqual(serialized_collectors_item["description"], "")
self.assertEqual(serialized_collectors_item["collectors_item_type"], collectors_item_dbo.collectors_item_type)
self.assertEqual(serialized_collectors_item["date_added"], str(date_added))
Notice that test_serialize_collectors_item is similar to test_create_collectors_item. In fact, we expand on the collectors_item creation test and we call serialize() on the collectors_item_dbo object after it has been saved to the database so that we can compare all the properties in there. The second collectors_item object being added is to test what happens when the nullable field description is left to None so we have complete test coverage of the tests/collectors_items/models/test_collectors_item_models.py file.
Running tests
To run the tests, we have to add a configuration file for the code coverage tool in the root of the project:
collectors_items-at-rest + .coveragerc
We will use this file to configure exclusion of the tests folder for code coverage. To continue, add the following lines in .coveragerc file:
[run] omit=tests*
We can now run the pytest-watcher tool with the following command:
ptw -- --cov-report term:skip-covered --cov=app --cov-config=.coveragerc tests
This command will produce a new test and test coverage report in the terminal.
We can now run automated testing of parts of our flask REST API.
Schema tests
Next, we will write tests for the schemas. This is a great time to use parametrization functionality from pytest. Parameterization allows us to write a single test for multiple input scenarios.
We will start with testing create_collectors_item_schema, first, the imports:
from datetime import datetime from jsonschema import Draft7Validator as d7val import pytest from app.main.collectors_item.schemas.collectors_item_schema import ( get_collectors_item_schema, create_collectors_item_schema, get_collectors_items_schema, update_collectors_item_schema, )
These are the required imports for testing the schemas. The Draft7Validator object will allow us to validate data using the json schemas that we defined in our project. The import for pytest will be used to parametrize our tests.
test_collectors_item_schemas:
@pytest.mark.parametrize(
"create_data",
[
(
{
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
}
),
(
{
"name": "test1",
"collectors_item_type": "testtype1",
}
),
],
)
def test_valid_create_collectors_item_schema(create_data):
d7val(create_collectors_item_schema).validate(create_data)
Here we define a couple of sets of input data. Since “description” should be optional, we test the schema validation with and without this property to verify that both data structures are valid according to the schema. We don’t need to use an assert statement here since the schema validator will raise an exception if the data does not match the schema, which will cause the test to fail.
In the next test scenario, we will verify that “name” and “collectors_item_type” are required properties:
@pytest.mark.parametrize(
"create_data",
[
(
{
"description": "test creation",
"collectors_item_type": "testtype1",
}
),
(
{
"name": "test1",
"description": "test creation",
}
),
(
{
"description": "test creation",
}
),
],
)
def test_invalid_create_collectors_item_schema(create_data):
assert d7val(create_collectors_item_schema).is_valid(create_data) is False
Here we do use an assert statement and use the is_valid function on the validator to get a boolean return value to evaluate.
The test for the update schema follows a similar pattern, but since everything is optional, there is no way to actually invalidate it.
The best practice is to verify that all the combinations are possible:
...
@pytest.mark.parametrize(
"update_data",
[
(
{
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
}
),
(
{
"name": "test1",
"collectors_item_type": "testtype1",
}
),
(
{
"name": "test1",
}
),
(
{
"collectors_item_type": "testtype1",
}
),
(
{
"description": "test creation",
}
),
],
)
def test_valid_update_collectors_item_schema(update_data):
d7val(update_collectors_item_schema).validate(update_data)
Next is a test for the get_collectors_items_schema:
@pytest.mark.parametrize(
"get_data",
[
(
[
{
"id": 1,
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
"date_added": str(datetime.utcnow()),
},
{
"id": 2,
"name": "test2",
"description": "test creation2",
"collectors_item_type": "testtype2",
"date_added": str(datetime.utcnow()),
},
]
),
([]),
],
)
def test_valid_get_collectors_items_schema(get_data):
d7val(get_collectors_items_schema).validate(get_data)
There are two parameters for this test. The first is an array with two collectors_items. All the required fields are present. The second one is an empty array, which should also be a valid response.
The following code tests the invalid scenarios:
@pytest.mark.parametrize(
"get_data",
[
(
[
{
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
"date_added": str(datetime.utcnow()),
},
{
"id": 2,
"name": "test2",
"description": "test creation2",
"collectors_item_type": "testtype2",
"date_added": str(datetime.utcnow()),
},
]
),
(
[
{
"id": 1,
"description": "test creation",
"collectors_item_type": "testtype1",
"date_added": str(datetime.utcnow()),
},
]
),
(
[
{
"id": 1,
"name": "test1",
"collectors_item_type": "testtype1",
"date_added": str(datetime.utcnow()),
},
]
),
(
[
{
"id": 1,
"name": "test1",
"description": "test creation",
"date_added": str(datetime.utcnow()),
},
]
),
(
[
{
"id": 1,
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
},
]
),
],
)
def test_invalid_get_collectors_items_schema(get_data):
assert d7val(get_collectors_items_schema).is_valid(get_data) is False
The first test parameter has two items in the array to test if the data is still invalid even if one of the items is correct. Which should obviously invalidate the whole data set, but it is good to check even the obvious scenarios.
Endpoint tests
Last but not least in our effort of implementing automated testing of our flask REST API, we will write tests for the endpoints (get, post, put, and delete).
These tests will be like integration tests. They will execute actual HTTP requests to the API endpoints.
tests/collectors_items/controllers/test_collectors_item_controllers.py
The first test we will set up here, will test the “create collectors items” endpoint.
Create collectors items
Below is the entire code to set up these tests. We will go through them section by section.
import json
from tests.base import BaseTestCase
def create_collectors_item(self, collectors_item_data):
return self.client.post(
"/collectors_items",
data=json.dumps(collectors_item_data),
content_type="application/json",
)
class TestCollectorsItemController(BaseTestCase):
def test_create_collectors_items(self):
collectors_item_data = {
"name": "test1",
"description": "test creation",
"collectors_item_type": "testtype1",
}
with self.client:
response = create_collectors_item(self, collectors_item_data)
self.assertEqual(response.status_code, 201)
actual_result = json.loads(response.data.decode())
self.assertEqual(actual_result["id"], 1)
self.assertEqual(actual_result["name"], "test1")
self.assertEqual(actual_result["description"], "test creation")
self.assertEqual(actual_result["collectors_item_type"], "testtype1")
Helper function
The create_collectors_item function is used to make a HTTP request to the running flask app in the test scenario. With this HTTP request we can test the endpoint and service code,
from start to finish. The self parameter here will be the self from the test class derived from BaseTestCase which in turn is derived from the flask_testing.TestCase class. This class contains client which can execute HTTP requests. In the case of creating an item, we do a post request. The first parameter indicates the endpoint address, then the data payload, and finally the content_type.
Test class
We then create a class derived from BaseTestCase as we did for previous tests. This will ensure the flask app and database are set up for the test scenarios.
The function test_create_collectors_items describes the actual test. With collectors_item_data we represent data that would be sent by a client to our endpoint. In a previous tutorial, we created data schemas for our endpoints that validate the data coming in from clients, so we have to make sure the data we’re sending conforms to the schema for this endpoint.
We will place the call in a with self.client: block to make sure client resources are cleaned up after the test is done.
We call the create_collectors_item function with the appropriate parameters and then examine the response we get back. First, it is important to look at the status_code because from the status_code we can instantly learn if something went wrong. If something in the flask app caused an exception, we will probably get a status code 500 for example. If the status_code is ok, we can examine the data we received as a response. We decode and load the response data as json with actual_result = json.loads(response.data.decode())
, then we verify all the expected properties.
The pattern of “execute HTTP request, verify status code, verify response data” will repeat for all the other endpoint tests as well.
Get collectors items
The next endpoint we will test is the get endpoint to retrieve collectors_items from the app.
We have to set up a scenario where there is data to retrieve through the endpoint. To do that, we need to add imports for database operations.
test_collectors_item_controllers:
... from app.main.collectors_item.models import collectors_item_model from app.main import _db
The following code is used to put data into the database for the test scenarios that require it. Place it somewhere above the test class TestCollectorsItemController:
def create_dummy_collectors_items(date_added):
collectors_item_dbo = collectors_item_model.CollectorsItemModel(
name="test_collectors_item",
description="collectors_item for testing",
collectors_item_type="ccc2",
date_added=date_added,
)
_db.session.add(collectors_item_dbo)
_db.session.commit()
Here we create a CollectorsItemModel object and commit it to the database.
We need a HTTP request function for get. Place the following code below create_dummy_collectors_items:
...
def get_collectors_items(self):
return self.client.get("/collectors_items", content_type="application/json")
...
This code is similar to create_dummy_collectors_items, except that it doesn’t send data.
Finally we add a new test:
...
def test_get_collectors_items(self):
date_added = datetime.utcnow()
create_dummy_collectors_items(date_added)
with self.client:
response = get_collectors_items(self)
self.assertEqual(response.status_code, 200)
expected_result = [
{
"id": 1,
"name": "test_collectors_item",
"description": "collectors_item for testing",
"collectors_item_type": "ccc2",
"date_added": str(date_added),
}
]
actual_result = json.loads(response.data.decode())
self.assertEqual(actual_result, expected_result)
We call the previously created create_dummy_collectors_items function to populate the database for this scenario. Then, we call get_collectors_items to request a response.
We check the status_code first and then compare the actual result with the expected result. We’re expecting a list with one entry.
Update collectors item
To test the “update collectors item” functionality, we need to execute a put HTTP request. We are targeting a specific item with the update so the endpoint address should include the id for the item we are updating:
...
def update_collectors_item(self, collectors_item_id, collectors_item_data):
return self.client.put(
"/collectors_items/{}".format(collectors_item_id),
data=json.dumps(collectors_item_data),
content_type="application/json",
)
...
The test itself looks like this:
...
def test_update_collectors_items(self):
date_added = datetime.utcnow()
create_dummy_collectors_items(date_added)
with self.client:
update_item_data = {
"name": "updated_name",
"description": "updated description",
"collectors_item_type": "updated_type",
}
response = update_collectors_item(self, 1, update_item_data)
self.assertEqual(response.status_code, 200)
actual_result = json.loads(response.data.decode())
self.assertEqual(actual_result["id"], 1)
self.assertEqual(actual_result["name"], "updated_name")
self.assertEqual(actual_result["description"], "updated description")
self.assertEqual(actual_result["collectors_item_type"], "updated_type")
...
Not much new here. We populate the database again with create_dummy_collectors_items and then send the update request for collectors item with id 1. Because there will be only 1 item in the database and its id will be 1.
Delete collectors item
To test the “delete collectors item” functionality, we need to execute a delete HTTP request. We are targeting a specific item to delete so the endpoint address should include the id for the item we are deleting:
...
def delete_collectors_item(self, collectors_item_id):
return self.client.delete(
"/collectors_items/{}".format(collectors_item_id),
content_type="application/json",
)
...
The actual test:
...
def test_delete_collectors_item(self):
date_added = datetime.utcnow()
create_dummy_collectors_items(date_added)
with self.client:
response = delete_collectors_item(self, 1)
self.assertEqual(response.status_code, 200)
actual_result = json.loads(response.data.decode())
self.assertEqual(actual_result["status"], "success")
Again the same basic pattern as in the previous tests for the endpoints applies here.
We have now written basic tests for all the endpoints. These are all “good” or “expected” flows. Let us add some examples of scenarios where something goes wrong.
Error handling
For example, the scenario where the client tries to update an item that has been deleted by a different client:
...
def test_update_collectors_items_no_items(self):
date_added = datetime.utcnow()
with self.client:
update_item_data = {
"name": "updated_name",
"description": "updated description",
"collectors_item_type": "updated_type",
}
response = update_collectors_item(self, 1, update_item_data)
self.assertEqual(response.status_code, 404)
actual_result = json.loads(response.data.decode())
self.assertEqual(actual_result["status"], "failed")
...
Unlike the other “update collectors items” scenario, here we do not add an item to the database first. The database remains empty. The update should fail and return a 404, resource not found, error.
Conclusion
We have now learned automated testing for our flask REST API . With these tests, we are testing important aspects of the REST API project we had set up in the tutorial: Dockerized Flask RESTful API: how to. The completed project can be found in the following github repository: https://github.com/tmsdev82/collectors-items-at-rest.
Integration or Unit tests
A note about the types of tests we’ve written, most of the tests we’ve written are technically “integration” tests. Tests that include communication with external systems. For example: a database or a client sending HTTP requests to our endpoint. The external system in this case is an in memory database and a client we have control over.
Database tests
Since the database is in memory and recreated as completely clean for each test scenario, there is no risk of a database with unknown data that could influence the tests. There also are no driver issues that could fail the tests. Furthermore, the speed of testing is high because the database is in memory.
Endpoint tests
The same is true for the HTTP requests we did for our endpoint tests. The client is an “internal” system that sends the requests, so very little can go wrong in the communication there. We have direct control over what the client sends as well, we can be confident that those factors cannot influence the results of the tests either.
Schema tests
The tests for the schema validation are proper unit tests. Small, concise and not touching any external systems.
For more information on testing in python, try these articles:
Information on continuous testing: Continuous testing python code with pytest: how to.
Information about mocking and parametrization in tests can be found in the article: Mocking in python tests: practical examples.