You are currently viewing Part 6: Unit testing your microservice – Ensuring reliability and stability with pytest

Part 6: Unit testing your microservice – Ensuring reliability and stability with pytest

  • Post author:
  • Post category:Python

Welcome to Part 6 of the tutorial series: “Build a Production-Ready User Management Microservice with Flask and SQLAlchemy: A Step-by-Step Guide”. In this part, we’ll focus on writing unit tests for your Flask application. Unit testing ensures your microservice is reliable, maintainable, and free of bugs. By the end of this tutorial, you’ll have a fully tested Flask microservice, giving you confidence in its functionality and stability.

What you learn in part 6

  • Understand the importance of unit testing for building robust applications.
  • Set up a testing environment for your Flask application.
  • Write unit tests for routes, models, and authentication logic.
  • Test edge cases and error scenarios.
  • Run and automate tests using tools like pytest and unittest.

Prerequisites

Before you begin, complete Part 1Part 2Part 3Part 4, and Part 5 of the series. You should have:

  • A Flask application with SQLAlchemy configured.
  • User model with password hashing.
  • CRUD operations for the User model.
  • JWT authentication for securing endpoints.
  • Error handling and input validation.
  • Logging and monitoring configured.

Step 1: Set up a testing environment

To write unit tests, configure a separate testing environment. This ensures tests don’t interfere with your development or production databases.

Install testing dependencies

Install the following Python packages for testing:

pip install pytest pytest-cov requests
  • pytest: A testing framework for writing and running tests.
  • pytest-cov: A plugin for measuring test coverage.
  • requests: A library for making HTTP requests (useful for testing API endpoints).

Update app/__init__.py

Modify the create_app function to support a testing configuration:

import os
import logging
from logging.handlers import RotatingFileHandler
from prometheus_flask_exporter import PrometheusMetrics
from flask import Flask, jsonify, request
from flask_sqlalchemy import SQLAlchemy
from flask_jwt_extended import JWTManager
from config import Config

db = SQLAlchemy()
jwt = JWTManager()

def create_app(test_config=None):
    app = Flask(__name__)

    if test_config is None:
        app.config.from_object('config.Config')
    else:
        app.config.from_mapping(test_config)

    db.init_app(app)
    jwt.init_app(app)

    # Initialize Prometheus metrics
    metrics = PrometheusMetrics(app)  # Enable default metrics

    # Configure logging
    if not app.debug:
        # Create logs directory if it doesn't exist
        if not os.path.exists('logs'):
            os.mkdir('logs')

        # Set up a rotating file handler
        file_handler = RotatingFileHandler(
            'logs/microservice.log', maxBytes=10240, 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)

        # Set the log level for the application
        app.logger.setLevel(logging.INFO)
        app.logger.info('Microservice startup')

    @app.errorhandler(404)
    def not_found(error):
        return  jsonify({"error": "Not Found", "message": "The requested resource was not found"}), 404
    
    @app.errorhandler(500)
    def internal_error(error):
        return jsonify({"error": "Internal Server Error", "message": "Something went wrong on the server"}), 500

    with app.app_context():
        from . import routes
        app.register_blueprint(routes.bp)

        db.create_all()

    return app

The code configures unit testing by allowing the app to load a custom test configuration. When test_config is provided, the app overrides the default configuration using app.config.from_mapping(test_config). This ensures tests run with specific settings, such as an in-memory database, without modifying the production configuration.

Create a test configuration

Add a test configuration to your config.py file:

import os

class Config:
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL', 'sqlite:///app.db')  # Use SQLite by default
    SQLALCHEMY_TRACK_MODIFICATIONS = False  # Disable modification tracking
    JWT_SECRET_KEY = os.getenv('JWT_SECRET_KEY', 'your-secret-key')  # Change this to a secure key
    DEBUG_METRICS=True

class TestConfig(Config):
    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:' # Use an in-memory SQLite database for testing
    TESTING = True

The code configures unit testing by defining TestConfig, which inherits from Config. It sets SQLALCHEMY_DATABASE_URI to an in-memory SQLite database, ensuring tests run in isolation without affecting real data. It also enables TESTING, which allows Flask to handle errors differently and provides better debugging during tests.

Directory Structure

your_project/
├── app/
│   ├── __init__.py
│   ├── routes.py
│   ├── models.py
│   └── ...
├── tests/
│   ├── __init__.py
│   ├── conftest.py    
│   ├── test_models.py
│   ├── test_routes.py
│   └── ...
├── config.py
├── logs/
├── docker-compose.yml
├── prometheus.yml
└── ...

The tests/ folder organizes unit and integration tests for the project.

  • __init__.py: Marks the folder as a Python package, allowing test modules to be imported.
  • conftest.py: Defines reusable test fixtures (e.g., creating a test app and database) to simplify test setup.
  • test_models.py: Contains tests for database models, ensuring data validation and relationships work correctly.
  • test_routes.py: Tests API routes to verify request handling, authentication, and expected responses.

By structuring tests this way, the project ensures reliability and maintainability while keeping test logic separate from application code.

Step 2: Create conftest.py

The conftest.py file defines fixtures that you can share across multiple test files. This avoids code duplication and makes your test suite more maintainable.

Create conftest.py

Create a conftest.py file in the tests directory and add the following code:

import pytest
from app import create_app, db

@pytest.fixture
def app():
    # Create the Flask app with test configuration
    app = create_app(test_config={
        'SQLALCHEMY_DATABASE_URI': 'sqlite:///:memory:',
        'TESTING': True,
        'JWT_SECRET_KEY': 'test-secret-key'  # Add JWT secret key for testing
    })
    
    # Set up the application context and database
    with app.app_context():
        db.create_all()  # Create all database tables
        yield app        # Yield the app for testing
        db.drop_all()    # Drop all database tables after testing

@pytest.fixture
def client(app):
    # Create a test client for making requests
    return app.test_client()

The code sets up unit testing for a Flask application using pytest. The app fixture creates the Flask app with a test configuration, initializes the database, and ensures a clean state by creating and dropping tables before and after tests. The client fixture provides a test client to simulate HTTP requests, enabling API testing without running the server.

Step 3: Write unit tests

Write unit tests for the following:

  1. Database Models: Test the User model and its methods.
  2. Routes: Test API endpoints (e.g., /api/register/api/login).
  3. Authentication: Test JWT authentication and protected routes.

Test database models

Create a file tests/test_models.py to test the User model:

import pytest
from app import db
from app.models import User

@pytest.fixture

def test_user_creation(app):
    with app.app_context():
        user = User(username='testuser', email='test@example.com')
        user.set_password('password')
        db.session.add(user)
        db.session.commit()

        # Retrieve the user from the database
        retrieved_user = User.query.filter_by(username='testuser').first()
        assert retrieved_user is not None
        assert retrieved_user.email == 'test@example.com'
        assert retrieved_user.check_password('password') is True
        assert retrieved_user.check_password('wrongpassword') is False

This code defines a test for creating and validating a user in the database.

  • The test_user_creation function runs within the test application context.
  • It creates a User instance with a username and email, then sets the password securely.
  • The function adds the user to the database and commits the transaction.
  • It retrieves the user by username and verifies that the user exists.
  • It asserts that the retrieved email matches the expected value.
  • It checks that the correct password returns True, while an incorrect password returns False.

This test ensures that user creation, password hashing, and authentication work correctly.

Test Routes

Create a file tests/test_routes.py to test API endpoints. Here’s the updated code for test_routes.py:

def test_register_user(client):
    response = client.post('/api/register', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'password'
    })
    assert response.status_code == 201
    assert 'id' in response.json
    assert response.json['username'] == 'testuser'
    assert response.json['email'] == 'test@example.com'

def test_login_user(client):
    # Register a user first
    register_response = client.post('/api/register', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'password'
    })
    assert register_response.status_code == 201  # Ensure registration is successful

    # Test login with correct credentials
    login_response = client.post('/api/login', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'password'
    })
    assert login_response.status_code == 200  # Ensure login is successful
    assert 'access_token' in login_response.json  # Ensure access_token is present

    # Test login with incorrect credentials
    failed_login_response = client.post('/api/login', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'wrongpassword'
    })
    assert failed_login_response.status_code == 401  # Ensure login fails
    assert 'error' in failed_login_response.json  # Ensure error message is present

def test_protected_route(client):
    # Register a new user
    client.post('/api/register', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'password'
    })

    # Log in to get a JWT token
    login_response = client.post('/api/login', json={
        'username': 'testuser',
        'email': 'test@example.com',
        'password': 'password'
    })

    assert login_response.status_code == 200  # Ensure login is successful

    # Extract the access token
    token = login_response.get_json().get("access_token")
    assert token is not None  # Ensure token was returned

    # Include the token in the Authorization header
    headers = {"Authorization": f"Bearer {token}"}

    # Correct route: "/api/protected"
    response = client.get('/api/protected', headers=headers)
    assert response.status_code == 200  # Should return 200 if authorized
    assert 'message' in response.get_json()  # Ensure response contains a message

    assert 'message' in response.get_json()  # Ensure response contains a message

This code defines unit tests for user registration, login, and access to a protected route.

  • test_register_user
    • Sends a POST request to /api/register with user details.
    • Asserts that the response status code is 201, indicating successful registration.
    • Ensures that the response contains the user’s id, username, and email.
  • test_login_user
    • Registers a user first to ensure login has valid credentials.
    • Sends a POST request to /api/login with correct credentials and verifies a 200 status and the presence of an access_token.
    • Attempts to log in with an incorrect password and asserts that the response returns 401 with an error message.
  • test_protected_route
    • Registers a user and logs in to obtain an access_token.
    • Extracts the token from the login response and includes it in the Authorization header as Bearer <token>.
    • Sends a GET request to /api/protected and asserts that the response returns 200 with a valid message.

These tests verify that user authentication, authorization, and access control work correctly.

Step 4: Run and automate tests

Run tests

Run the tests using pytest:

pytest tests/ --cov=app

Expected output

When you run the command, you’ll see output similar to this:

============================= test session starts =============================
platform darwin -- Python 3.12.3, pytest-8.3.4, pluggy-1.5.0
rootdir: /path/to/your_project
plugins: cov-6.0.0
collected 3 items

tests/test_routes.py ...                                                [100%]

---------- coverage: platform darwin, python 3.12.3-final-0 ----------
Name              Stmts   Miss  Cover
-------------------------------------
app/__init__.py      38      4    89%
app/models.py        13      1    92%
app/routes.py        85     31    64%
app/schemas.py        6      0   100%
-------------------------------------
TOTAL               142     36    75%

============================== 3 passed in 0.35s ==============================

Description of the output

  1. Test session summary:
    • The output starts with details about the test environment, including the Python version, pytest version, and the directory where the tests are running.
    • It lists the number of tests collected (e.g., collected 3 items).
  2. Test progress:
    • Each dot (.) represents a passing test. For example:
      • tests/test_routes.py ... indicates three tests passed in test_routes.py.
  3. Coverage Report:
    • The --cov=app flag generates a coverage report showing how much of your code the tests cover.
    • The report includes:
      • Stmts: Total number of executable statements in the file.
      • Miss: Number of statements not executed by the tests.
      • Cover: Percentage of code covered by tests.
    • In the example above:
      • app/__init__.py has 89% coverage.
      • app/models.py has 92% coverage.
      • app/routes.py has 64% coverage.
      • app/schemas.py has 100% coverage.
      • The TOTAL coverage is 75%.
  4. Test Summary:
    • The final line (3 passed in 0.35s) confirms that all 3 tests passed successfully in 0.35 seconds.

Automate tests

Integrate tests into a CI/CD pipeline (e.g., GitHub Actions, GitLab CI). Here’s an example GitHub Actions workflow (.github/workflows/tests.yml):

name: Run Tests

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      - name: Set up Python
        uses: actions/setup-python@v2
        with:
          python-version: '3.9'
      - name: Install dependencies
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
          pip install pytest pytest-cov
      - name: Run tests
        run: pytest tests/ --cov=app

This GitHub Actions workflow runs tests automatically whenever someone pushes code or opens a pull request. It first checks out the repository, then sets up Python 3.9 on the runner. Next, it upgrades pip, installs dependencies from requirements.txt, and adds pytest and pytest-cov for testing and coverage reporting. Finally, it runs pytest on the tests/ directory while measuring code coverage for the app module. This workflow ensures that all changes pass the test suite before merging, improving code quality and stability.

Task: write more tests for remaining endpoints

Your current test coverage for app/routes.py is 64%, which means there are still untested endpoints. Your task is to write additional tests to improve the coverage. Here’s what you need to do:

1. Test the /api/users Endpoint

  • Write tests for the following scenarios:
    • Fetch all users (GET /api/users).
    • Fetch a single user by ID (GET /api/users/<id>).
    • Update a user (PUT /api/users/<id>).
    • Delete a user (DELETE /api/users/<id>).

2. Test error scenarios

  • Write tests for error cases, such as:
    • Fetching a non-existent user (GET /api/users/999).
    • Updating a user with invalid data (PUT /api/users/<id> with missing fields).
    • Deleting a non-existent user (DELETE /api/users/999).

3. Test protected routes

  • Write tests for protected routes, such as:
    • Accessing /api/protected without a valid JWT token.
    • Accessing /api/protected with an expired or invalid token.

4. Test edge cases

  • Write tests for edge cases, such as:
    • Registering a user with a duplicate username or email.
    • Logging in with incorrect credentials multiple times (test rate limiting, if implemented).

5. Improve coverage for app/__init__.py

  • Write tests for the create_app function to ensure it works correctly with and without a test_config.

Full code for part 5

You can find the complete code for this tutorial in the GitHub repository.

What’s Next?

In Part 7, you’ll dive into containerizing your microservice with Docker. Here’s what you’ll learn:

  • Create a Dockerfile for your Flask application.
  • Use Docker Compose to manage multi-container setups.
  • Run your application in a containerized environment.

Stay tuned! 🚀

Facebook Comments