Introduction
In this article, we are going to be learning about how to write tests for our flask applications. Software testing is important so as to catch errors and unexpected behaviours in our code before pushing them to production. In this article, we are going to be building a simple note-taking application and we are going to write tests for it. For this article, make sure you have python3 and MongoDB installed on your local development machine.
Setup
Before we begin, we have to setup virtualenv for the project so as to keep the packages used in the application isolated. To do this, open your terminal in the directory you want to use and execute the following commands:
mkdir notes && cd notes
mkdir src tests
cd src
mkdir common controllers services && touch app.py
python3 -m venv venv
The above commands create your project folder the virtualenv. The command on line 3 is the one responsible for creating the virtualenv. Your folder directory should look like this after executing the commands below:
|-- notes/
|-- src/
|-- common/
|-- services/
|-- controllers/
|-- app.py
|-- tests/
|-- venv/
Run the commands below to activate the virtualenv created.
source venv/bin/activate
Required libraries:
The flask library is used to create the webserver while the pymongo library is used to connect to mongodb and perform mongodb operations. The pytest library will be used to write and run our tests.
Run the commands below to install the required libraries:
pip install Flask pymongo pytest
Building the Application
First of all, we are going to write the logic for our database operations i.e. creating notes, retrieving notes, updating notes and deleting notes. Create a file called notes.py
in the services
folder and copy the following inside.
import os
from bson.objectid import ObjectId
from pymongo import MongoClient
mongodb_uri = os.getenv('MONGODB_URI', 'mongodb://localhost/')
db_name = os.getenv('DB_NAME', 'notes')
client = MongoClient(mongodb_uri)
db = client[db_name]
notes = db['notes']
def create_note(data):
notes.insert_one(data)
def update_note(note_id, data):
note = notes.find_one_and_update({'_id': ObjectId(note_id)}, {'$set': data}, return_document=True)
return note
def delete_note(note_id):
notes.find_one_and_delete({'_id': ObjectId(note_id)})
def get_notes(conditions):
if '_id' in conditions:
conditions['_id'] = ObjectId(conditions['_id'])
results = notes.find(conditions)
notes_data = []
for data in results:
notes_data.append(dict(data))
return notes_data
In the above code, we import MongoClient
from the pymongo library to connect to the MongoDB instance as specified in the environment variables. We also set it to connect to our local MongoDB instance if no environment variable is provided. The function get_notes
accepts conditions to use to query for data in our database. It checks if _id
field is part of the fields being used to query and if so, converts it from a string to an ObjectId
object because the _id
field is not stored as a string in MongoDB but an ObjectId
instance. Also, note that the notes.find
function returns a cursor instance and we have to iterate over it to get our required data. You can read more about this here.
Next, we are going to write the utility functions for returning api responses. Create a file called utils.py
in the common
folder and copy the following inside:
import time
import json
from flask import jsonify
def stringify_objectid(data):
str_data = json.dumps(data, default=str)
return json.loads(str_data)
def response(status, message, data, status_code=200):
"""
:param status : Boolean Status of the request
:param status_code: Status Code of response
:param message : String message to be sent out as description of the message
:param data : dictionary representing extra data
"""
if data:
data = stringify_objectid(data)
res = {'status': status, 'message': message, 'data': data, 'timestamp': timestamp()}
return jsonify(res), status_code
def error_response(message, status='error', code='R0', status_code=400):
res = {'message': message, 'status': status, 'code': code}
return jsonify(res), status_code
def timestamp():
"""
Helper Function to generate the current time
"""
return time.time()
def validate_body(data, required_fields):
for field in required_fields:
if field not in data:
return False, field
return True, None
The stringify_objectid
function is for converting the _id
field from an ObjectId
instance to a string. The function json.dumps(data, default=``str``)
converts the data sent to a string and if it encounters any field that it can’t decode, it forcibly converts it to a string by calling the str
function on it. This makes sure that our _id
field is sent back as a string.
The validate_body
function checks if the data passed as all the required fields and returns a boolean status based on whether a field is missing or not.
Next, let us write our controllers, create a file called notes.py
in the controllers
folder and copy the following inside.
from flask import Blueprint, request
from src.common.utils import response, error_response, validate_body
from src.services.notes import create_note, get_notes, update_note, delete_note
notes = Blueprint('notes', __name__)
@notes.route('/', methods=['POST'])
def create():
body = request.get_json()
status, missing_field = validate_body(body, ['title', 'description'])
if not status:
return error_response(f'{missing_field} is required')
try:
create_note(body) # _id is automatically added to body
return response(True, 'Note created successfully', body)
except Exception as err:
print(f'::::: Error', err)
return error_response(str(err))
@notes.route('/', methods=['GET'])
def view():
conditions = dict(request.args)
try:
data = get_notes(conditions)
return response(True, 'Notes', data)
except Exception as err:
print(f'::::: Error', err)
return error_response(str(err))
@notes.route('/<note_id>', methods=['PUT'])
def update(note_id):
body = request.get_json()
try:
note = update_note(note_id, body)
return response(True, 'Note updated successfully', note)
except Exception as err:
print(f'::::: Error', err)
return error_response(str(err))
@notes.route('/<note_id>', methods=['DELETE'])
def delete(note_id):
try:
delete_note(note_id)
return response(True, 'Note deleted successfully', None)
except Exception as err:
print(f'::::: Error', err)
return error_response(str(err))
These are functions responsible for handling creating notes, updating notes, getting notes and deleting notes. The make use of the service functions and the utilities functions we created earlier.
Finally, open your app.js
file and copy the following inside:
from flask import Flask
from src.controllers.notes import notes
app = Flask(__name__)
app.register_blueprint(notes, url_prefix='/notes')
if __name__ == '__main__':
app.run()
The register_blueprint
function make sure that url starting with /notes
will be processed by the notes controller we wrote earlier. You can execute the app.py
file to run the application and test the endpoints written.
Testing the application
In this section, we are going to be writing tests for the logic written earlier. We are going to be using the pytest library to run the tests. First of all, we are going to be writing tests for the notes creation logic. Note that the names of our test files must start with test
. Create a file called test_create_note.py
in the tests directory and copy the following inside:
from unittest.mock import patch
import pytest
from src.app import app
@pytest.fixture
def client():
app.config['TESTING'] = True
with app.test_client() as client:
yield client
@pytest.fixture
def mock_insert():
with patch('pymongo.collection.Collection.insert_one') as mock_insert:
yield mock_insert
def test_insert_note_with_incomplete_fields(client):
rv = client.post('/notes/', json={'title': 'Test'})
json_data = rv.get_json()
assert 'description is required' == json_data['message']
def test_insert_note(client, mock_insert):
note_data = {'title': 'Test', 'description': 'Test Description'}
rv = client.post('/notes/', json=note_data)
json_data = rv.get_json()
mock_insert.assert_called_with(note_data)
assert 'Test' == json_data['data']['title']
Fixtures are functions that are to be executed before each test case is run. If we want to use the value returned from the test fixtures in our test functions, we just add a parameter with the same name as the fixture function to our function. Note that our test functions must start with test_
for pytest to pick it up. The client
fixture function creates returns a test client we can use to mimic API calls in our tests. The mock_insert
fixture function patches the insert_one
functionality of the pymongo
library and makes sure that when the insert_one
function is called in our service function, it’s the mock_insert
function that is called. We are going to use this test that the code calls the MongoDB insert functionality with the right arguments. The test_insert_note
function calls the /notes
endpoints and asserts that the mock_insert
function is called with the data sent. It also checks whether the right data is sent back in the JSON response.
Next, we are going to write tests for the update note functionality. Create a file called test_update_note.py
in the test
directory and copy the following inside:
from unittest.mock import patch
import pytest
from bson.objectid import ObjectId
from src.app import app
@pytest.fixture
def client():
app.config['TESTING'] = True
with app.test_client() as client:
yield client
@pytest.fixture
def mock_update():
with patch('pymongo.collection.Collection.find_one_and_update') as mock_update:
yield mock_update
def test_update_note(client, mock_update):
note_id = '5e13ca9d074eca4a9a9497c6'
input_data = {'title': 'Test 1'}
mock_update.return_value = {'_id': note_id, 'title': 'Test 1'}
rv = client.put(f'/notes/{note_id}', json=input_data)
json_data = rv.get_json()
mock_update.assert_called_with({'_id': ObjectId(note_id)}, {'$set': input_data}, return_document=True)
assert input_data['title'] == json_data['data']['title']
The mock_update
function patches the find_one_and_update
function used to update a specified document in the database. The test_update_note
function calls the notes update endpoint with test data and asserts that the database update function is called with the right arguments and the json data sent back as response contains the right data.
Next, we are going to write tests for the get notes functionality and the delete note functionality. Create a file called test_get_notes
in the tests
directory and copy the following inside:
from unittest.mock import patch
import pytest
from src.app import app
@pytest.fixture
def client():
app.config['TESTING'] = True
with app.test_client() as client:
yield client
@pytest.fixture
def mock_find():
with patch('pymongo.collection.Collection.find') as mock_find:
yield mock_find
def test_notes_view_with_no_conditions(client, mock_find):
mock_find.return_value = [{'title': 'Test'}, {'title': 'Test 1'}]
rv = client.get('/notes/')
json_data = rv.get_json()
mock_find.assert_called_with({})
assert len(json_data['data']) == 2
def test_notes_view_with_conditions(client, mock_find):
mock_find.return_value = [{'title': 'Test'}]
rv = client.get('/notes/?title=Test')
json_data = rv.get_json()
mock_find.assert_called_with({'title': 'Test'})
assert 'Test' == json_data['data'][0]['title']
assert len(json_data['data']) == 1
Create a file called test_delete_note.py
in the tests
directory and copy the following inside:
from unittest.mock import patch
import pytest
from bson.objectid import ObjectId
from src.app import app
@pytest.fixture
def client():
app.config['TESTING'] = True
with app.test_client() as client:
yield client
@pytest.fixture
def mock_delete():
with patch('pymongo.collection.Collection.find_one_and_delete') as mock_delete:
yield mock_delete
def test_delete_note(client, mock_delete):
note_id = '5e13ca9d074eca4a9a9497c6'
rv = client.delete(f'/notes/{note_id}')
json_data = rv.get_json()
mock_delete.assert_called_with({'_id': ObjectId(note_id)})
assert json_data['data'] is None
In the above code snippets, we patch the pymongo’s find
and find_one_and_delete
functions and test the endpoints associated them. We asserted that the mock functions were called and that the right JSON data is returned.
To run the above tests, run the command below:
pytest
You should see an output similar to what’s below in your terminal window:
======================================================================================= test session starts ========================================================================================
platform darwin -- Python 3.7.5, pytest-5.3.2, py-1.8.1, pluggy-0.13.1
rootdir: /Users/kudi/Workspace/notes
collected 6 items
tests/test_create_note.py .. [ 33%]
tests/test_delete_note.py . [ 50%]
tests/test_get_notes.py .. [ 83%]
tests/test_update_note.py . [100%]
======================================================================================== 6 passed in 0.39s =========================================================================================
Conclusion
In this article, we have learnt about how to test flask applications with pytest and mocking database functions with python unittest mock functionality. The full code for this article is available here .