Flask: Setting Up Tests and Logging.

Hello, welcome back to the 8th post of my Flask series, this post would be a continuation of my previous post, please click on any link below to view any post you missed

OUTLINE OF FLASK SERIES

• Structuring a Flask-Restful API for Production
• Creating Custom Error Pages and Handling Exceptions for Flask
• Flask authentication with JWT
• CRUD Operations with Flask
• Using Marshmallow to Simplify Parameter Validation in APIs.
• Email Setup, User Confirmation, and Password Reset Feature.
• Handling File Upload.
• Running tests and Flask Logging (this article)

At the end of this post, you will learn the following

• Set up unit tests.
• Run unit tests and check testing coverage
• Logging errors to a file and sending error logs via email.

We would install new dependencies for our new post

• coverage:  It measures code coverage, typically during test execution. It uses the code analysis tools and tracing hooks provided in the Python standard library to determine which lines are executable, and which have been executed.

Create a new database for testing

mysql -u root
mysql> CREATE DATABASE test_db;
Query OK, 1 row affected (0.00 sec)
mysql> GRANT ALL PRIVILEGES ON test_db . * TO 'diary_admin'@'localhost';
Query OK, 0 rows affected (0.00 sec)

Update your env file and add the database URL for testing

TEST_DATABASE_URL='mysql+pymysql://*******'

Our tests would be written using the standard unittest package from the Python standard library.

We would add the following code to our main.py in order to execute custom commands for our tests and to have our coverage metrics integrate into the flask test .

import os
import subprocess
import sys
import unittest

import click
import coverage
from dotenv import load_dotenv

from api.app import create_app

dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
if os.path.exists(dotenv_path):
    load_dotenv(dotenv_path)

app = create_app(os.getenv("FLASK_ENV"))

@app.cli.command()
def test():
    """Run the unit tests."""    
    tests = unittest.TestLoader().discover('tests')   
    unittest.TextTestRunner(verbosity=2).run(tests)

COV = None
if os.environ.get('FLASK_COVERAGE'):
    COV = coverage.coverage(branch=True, include=('resources/*','api/*'))
    COV.start()


@app.cli.command()
@click.option('--coverage/--no-coverage', default=False,
              help='Run tests under code coverage.')
@click.argument('test_names', nargs=-1)
def test(coverage, test_names):
    """Run the unit tests."""
    if coverage and not os.environ.get('FLASK_COVERAGE'):
        os.environ['FLASK_COVERAGE'] = '1'
        sys.exit(subprocess.call(sys.argv))
    if test_names:
        tests = unittest.TestLoader().loadTestsFromNames(test_names)
    else:
        tests = unittest.TestLoader().discover('tests')
    unittest.TextTestRunner(verbosity=2).run(tests)
    if COV:
        COV.stop()
        COV.save()
        print('Coverage Summary  of our app: ')
        COV.report()
        basedir = os.path.abspath(os.path.dirname(__file__))
        covdir = os.path.join(basedir, 'tmp/coverage')
        COV.html_report(directory=covdir)
        print('HTML version: file://%s/index.html' % covdir)
        COV.erase()

Update your config.py file too

class TestingConfig(Config):
    TESTING =  True
    # Disable CSRF protection in the testing configuration
    WTF_CSRF_ENABLED = False
    SQLALCHEMY_DATABASE_URI = os.environ.get(
    "TEST_DATABASE_URL"
    )

SET UP TESTS

mkdir tests
touch __init__.py test_user.py  test_note.py

Add the following to test our user resource, add the code snippets below to test_user.py

import datetime
import json
import os
import unittest
from unittest import TestCase
import io

from flask import url_for
from flask_jwt_extended import create_access_token

from api.app import create_app
from api.models import User, db
from utils.email_token import confirm_token, generate_confirmation_token

# registration data
payload = {
    "email": "oluchi@gmail.com",
    "password": "#hdyuER456*&",
    "username": "oluchidfg",
}
# login details
login_details = {
    "email": "oluchi@gmail.com",
    "password": "#hdyuER456*&",
}


class UserTestCase(TestCase):
    def setUp(self):
        self.app = create_app("testing")
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        self.client = self.app.test_client(use_cookies=True)

    def tearDown(self):
        """
        Will be called after every test
        """
        db.session.remove()
        db.drop_all()
        self.app_context.pop()

    def test_user_registration_login(self):

        # test user registration
        response = self.client.post(
            "/v1/user/register/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(payload),
        )

        self.assertEqual(response.status_code, 201)
        self.assertEqual(User.query.count(), 1)
        user = User.query.filter_by(username="oluchidfg").first()
        self.assertEqual(user.confirmed, False)
        self.assertEqual(user.confirmed_on, None)

        # Try to login as a user before confirmation

        response = self.client.post(
            "/v1/user/login/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(login_details),
        )
        self.assertTrue(
            "The user account is not activated yet" in response.get_data(as_text=True)
        )

        # now confirm the user
        user = User.query.filter_by(email="oluchi@gmail.com").first()
        token = generate_confirmation_token(user.email)
        response = self.client.get(
            "/users/activate/{}".format(token), follow_redirects=True
        )
        email = confirm_token(token)
        self.assertEqual(response.status_code, 200)
        self.assertTrue(
            "The user account has been activated" in response.get_data(as_text=True)
        )

        # test user after email confirmation
        response = self.client.post(
            "/v1/user/login/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(login_details),
        )
        res_data = response.json
        self.assertEqual(res_data["status"], "success")
        self.assertEqual(res_data["data"]["email"], "oluchi@gmail.com")
        self.assertEqual(res_data["data"]["user_id"], 1)
        self.assertEqual(response.status_code, 200)

        # test_login_with_invalid_password
        login_details["email"] = "oolluuu#@gmail.com"
        response = self.client.post(
            "/v1/user/login/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(login_details),
        )
        self.assertTrue(
            "username or password is incorrect" in response.get_data(as_text=True)
        )
        # try to register user with existing email
        data = {
            "email": "oluchi@gmail.com",
            "password": "#hdyuER456*&",
            "username": "oluchidfggggg",
        }
        response = self.client.post(
            "/v1/user/register/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(data),
        )
        res = response.json
        self.assertEqual(res["message"], "email already exist")
        self.assertEqual(response.status_code, 400)

        # try to register user with invalid email
        data = {
            "email": "oluchiiiigmail.com",
            "password": "#hdyuER456*&",
            "username": "oluchidfggggg",
        }
        response = self.client.post(
            "/v1/user/register/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(data),
        )
        res = response.json
        self.assertEqual(res["errors"]["email"], ["Not a valid email address."])
        self.assertEqual(response.status_code, 422)

        # recover password after forgetting login details
        email = payload["email"]

        token = generate_confirmation_token(email)

        password_reset = {
            "password": "hgRT45&*efg&",
            "confirmation_password": "hgRT45&*efg&",
        }
        email = confirm_token(token)
        response = self.client.patch(
            "/users/reset_password/{}".format(token),
            headers={"Content-Type": "application/json"},
            data=json.dumps(password_reset),
        )

        res = response.json
        self.assertEqual(res["data"]["msg"], "New password was successfully set")
        self.assertEqual(response.status_code, 200)

        # try to reset password with 2 umatching passwords
        email = payload["email"]

        token = generate_confirmation_token(email)

        password_reset = {
            "password": "hgRT45&*efg&",
            "confirmation_password": "hgRT45&fff*efg&",
        }
        email = confirm_token(token)
        response = self.client.patch(
            "/users/reset_password/{}".format(token),
            headers={"Content-Type": "application/json"},
            data=json.dumps(password_reset),
        )

        res = response.json
        self.assertEqual(res["message"], "The two password do not match")
        self.assertEqual(response.status_code, 400)

    def test_display_image_upload(self):
        user = User(
            username="oluchiiii",
            email="oluchi@gmail.com",
            password="xy%^FD345Yyuu*9",
            confirmed=True,
            confirmed_on=datetime.datetime.now(),
        )

        db.session.add(user)
        db.session.commit()
        
        

        access_token = create_access_token(identity=user.id, fresh=True)
        
        #test empty file upload
        data = {}
        response = self.client.put(
        '/v1/user/display_image/', 
        headers={
                "Content-Type": "multipart/form-data",
                "Authorization": f"Bearer {access_token}",
            },data=data,follow_redirects=True,

        )
        
        self.assertEqual(response.status_code, 400)

        #test  file upload
        filedata = {'file' : (io.BytesIO(b"image"), 'style3.jpg')}
        response = self.client.put(
        '/v1/user/display_image/', 
        headers={
                "Content-Type": "multipart/form-data",
                "Authorization": f"Bearer {access_token}",
            },data=filedata,follow_redirects=True,

        )
        res=response.json
        self.assertEqual(res["msg"], "uploaded display image successfully")
        self.assertEqual(response.status_code, 200)

Add the following to test our note resource, add the code snippets below to test_note.py

import datetime
import json
import os
import unittest
from unittest import TestCase

from api.app import create_app
from flask import url_for
from flask_jwt_extended import create_access_token
from api.models import User, db

# registration data


note_payload = {
    "title": "My Flask Testing",
    "notes": "I used Unittest LIbrary",
}


class NoteTestCase(TestCase):
    def setUp(self):
        self.app = create_app("testing")
        self.app_context = self.app.app_context()
        self.app_context.push()
        db.create_all()
        self.client = self.app.test_client(use_cookies=True)

    def tearDown(self):
        """
        Will be called after every test
        """
        db.session.remove()
        db.drop_all()
        self.app_context.pop()

    def test_notes(self):

        user = User(
            username="oluchiiii",
            email="oluchi@gmail.com",
            password="xy%^FD345Yyuu*9",
            confirmed=True,
            confirmed_on=datetime.datetime.now(),
        )

        db.session.add(user)
        db.session.commit()

        access_token = create_access_token(identity=user.id, fresh=True)
        res = self.client.post(
            "/v1/notes/",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
            data=json.dumps(note_payload),
        )

        results = res.json
        self.assertEqual(results["msg"], "successfully created notes")
        self.assertEqual(results["data"]["title_of_post"],  

        note_payload["title"])
        self.assertEqual(results["data"]["contents_of_post"], 

        note_payload["notes"])
        self.assertEqual(res.status_code, 201)

        # get draft notes
        res = self.client.get(
            "/v1/notes/draft/",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
        )

        self.assertEqual(res.status_code, 200)

        # #let's publish the note
        res = self.client.put(
            "/v1/publish_note/1",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
        )

        results = res.json
        self.assertEqual(results["msg"], "your note has been published succesfully")
        self.assertEqual(res.status_code, 200)

        # update note
        note_payload["title"] = "Going to try Pytest"
        res = self.client.put(
            "/v1/notes/1",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
            data=json.dumps(note_payload),
        )

        results = res.json
        self.assertEqual(results["msg"], "records updated successfully")
        self.assertNotEqual(results["data"]["title"], "My Flask Testing")
        self.assertEqual(res.status_code, 200)
        # delete  a note
        res = self.client.delete(
            "/v1/notes/1",
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {access_token}",
            },
        )
        results = res.json
        self.assertEqual(results["msg"], "action completed, note has been deleted")
        # write to create a note without authorization
        res = self.client.post(
            "/v1/notes/",
            headers={"Content-Type": "application/json"},
            data=json.dumps(note_payload),
        )

        results = res.json

        self.assertEqual(results["message"], "Missing Authorization Header")
        self.assertEqual(res.status_code, 401)

       



    

Run this on your terminal

flask test

test_notes (test_note.NoteTestCase) ... ok
test_display_image_upload (test_user.UserTestCase) ... ok
test_user_registration_login (test_user.UserTestCase) ... ok

----------------------------------------------------------------------
Ran 3 tests in 6.074s

OK

checking testing coverage

Test coverage is an important indicator of software quality and an essential part of software maintenance. It helps in evaluating the effectiveness of testing by providing data on different coverage items.

Run this on your terminal

flask flask test --coverage

test_notes (test_note.NoteTestCase) ... ok
test_display_image_upload (test_user.UserTestCase) ... ok
test_user_registration_login (test_user.UserTestCase) ... ok

----------------------------------------------------------------------
Ran 3 tests in 7.498s

OK
Coverage Summary  of our app: 
Name                 Stmts   Miss Branch BrPart  Cover
------------------------------------------------------
api/app.py              58     33      0      0    43%
api/models.py           59     47      2      0    20%
api/schemas.py          17     14      0      0    18%
resources/notes.py      74     42     20      7    44%
resources/user.py      147     87     38     10    43%
------------------------------------------------------
TOTAL                  355    223     60     17    39%
HTML version: file:///home/oluchi/Desktop/blog_Notes/diary_app/tmp/coverage/index.html

You notice a new folder tmp was created in your root directory, open the index.html to view on your browser.

We have a coverage of 38% which is not too bad but we can improve the result by writing more tests.

LOGGING A FLASK APP

Logging is a means of tracking events that happen when software runs. The software’s developer adds logging calls to their code to indicate that certain events have occurred. An event is described by a descriptive message which can optionally contain variable data (i.e. data that is potentially different for each occurrence of the event).

Logging is used to collect and store valuable information about the application’s execution, that can be used for troubleshooting (debugging) purposes. Logging includes either storing data to log files, sending it to monitoring services,  event logs, Unix Syslog, etc..

Logging provides a set of convenience functions for simple logging usage. These are debug(), info(), warning(), error() and critical()

The logging functions are named after the level or severity of the events they are used to track. The standard levels and their applicability are described below and arranged in increasing order of severity:

Handlers

Handler objects are responsible for dispatching the appropriate log messages (based on the log messages’ severity) to the handler’s specified destination.

• StreamHandler instances send messages to streams (file-like objects).
• FileHandler instances send messages to disk files.
• BaseRotatingHandler is the base class for handlers that rotate log files at a certain point. It is not meant to be instantiated directly.
• RotatingFileHandler instances send messages to disk files, with support for maximum log file sizes and log file rotation.
• TimedRotatingFileHandler instances send messages to disk files, rotating the log file at certain timed intervals.
• SocketHandler instances send messages to TCP/IP sockets.
• DatagramHandler instances send messages to UDP sockets.
• SMTPHandler instances send messages to a designated email address.

There are very few methods in a handler for application developers such as

• The setLevel() method, specifies the lowest severity that will be dispatched to the appropriate destination.
• setFormatter() selects a Formatter object for this handler to use and configures the final order, structure, and contents of the log message.                                                                                                      Example :

  logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')

  THE TABLE BELOW EXPLAINS  SOME IMPORTANT FORMATTER ATTRIBUTES

Let’s get into action.

Email Errors to Admins

When running the application on a remote server for production, you probably won’t be looking at the log messages very often. The WSGI server will probably send log messages to a file, and you’ll only check that file if a user tells you something went wrong.

To be proactive about discovering and fixing bugs, you can configure a logging.handlers.SMTPHandler to send an email when errors are logged.

Now update your Config file and add the following

import logging
from logging.handlers import SMTPHandler

from flask import current_app




#================existing code base

class ProductionConfig(Config):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = os.environ.get(
        "DATABASE_URL"
    ) or "sqlite:///" + os.path.join(basedir, "prod.sqlite")

    @classmethod
    def init_app(cls, app):
        Config.init_app(app)
        # email errors to the administrators

        credentials = (
            current_app.config["MAIL_USERNAME"],
            current_app.config["MAIL_PASSWORD"],
        )
        mailhost = (current_app.config["MAIL_SERVER"], current_app.config["MAIL_PORT"])
        secure = current_app.config["MAIL_USE_TLS"]
        fromaddr = current_app.config["MAIL_DEFAULT_SENDER"]
        ADMINS = ["oluchiexample@com"]
        subject = "Your Application Failed"

        mail_handler = SMTPHandler(
            mailhost=mailhost,
            fromaddr=fromaddr,
            toaddrs=ADMINS,
            subject=subject,
            credentials=credentials,
            secure=secure,
        )
        mail_handler.setLevel(logging.ERROR)
        mail_handler.setFormatter(
            logging.Formatter("[%(asctime)s] %(levelname)s in %(module)s: %(message)s")
        )
        app.logger.addHandler(mail_handler)

import logging
import os
from logging.handlers import SMTPHandler, RotatingFileHandler

from flask import current_app



class ProductionConfig(Config):
   #======================EXISTING CODE

        #log to a file
        if not os.path.exists('logs'):
            os.mkdir('logs')
        #limit the log size to 102MB
        file_handler = RotatingFileHandler('logs/diary_app.log', maxBytes=102400,
                                        backupCount=10)
        file_handler.setFormatter(logging.Formatter(
            '%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'))
        file_handler.setLevel(logging.INFO)
        app.logger.addHandler(file_handler)

        app.logger.setLevel(logging.INFO)
        app.logger.info('Starting our diary API')

Congratulations

I hope this has been as fun and interesting for you as it has for me! I’m looking forward to hearing your comments below.

References

• Logging

• Flask Web Development by Miguel Grinberg

Share this post: on Twitter on Facebook on Google+ on LinkedIn