Testing Flask Applications

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 .

No Comments Yet