This video is available to students only

How to Add Unit Tests to a Flask Stripe Payments Integration

Project Source Code

Get the project source code below, and follow along with the lesson material.

Download Project Source Code

To set up the project on your local machine, please follow the directions provided in the README.md file. If you run into any issues with running the project source code, then feel free to reach out to the author in the course's Discord channel.

Testing

Because payments are a core part of our product, it's an important component to test. There is a small hitch though, a lot of our pages rely on making (authenticated) network requests to Stripe's API. Our test suite should be able to be run offline and not depend on an external API. To solve that, we can mock the API response. We saw a primitive example of doing that early on where we forced requests.get to return a specific value.

A more robust way is to record what we expect network requests to return, and then use those in our test suite whenever a similar request is made. A library called "VCR.py" helps us do that, by allowing us to denote that certain tests should use "cassettes" of recorded network effects.

Add vcrpy to your requirements.txt and run pip install -r requirements.txt.

Our product details page now makes a request to Stripe's API. In order to capture what is being sent and received, we need to tell vcrpy to record the network requests made during the execution of that test.

Since we've already defined our Stripe configuration variables in our terminal environment, when we run the tests, they will actually make network requests to Stripe. vcrpy will observe these and write them into a folder.

Create a folder within yumroad/tests/ called cassettes. This is where we will store recorded requests. We will also want vcrpy to ignore sensitive authorization details so that we don't commit secrets into our codebase, so we can ask it to filter out authorization headers.

In tests/test_product.py, add a decorator to tell vcrpy to monitor this test. The record_once parameter will tell vcrpy to let the network request go through and store the output if there isn't already a recorded file.

import vcr
...
@vcr.use_cassette('tests/cassettes/new_stripe_session.yaml',
                  filter_headers=['authorization'], record_mode='once')
def test_details_page(client, init_database):
    book = create_book()
    response = client.get(url_for('product.details', product_id=book.id))
    assert response.status_code == 200
    assert b'Yumroad' in response.data
    assert b'Buy for $5.00' in response.data
    assert book.name in str(response.data)

Then once you confirm that your environment variables are configured correctly, you can run your test suite with pytest --cov-report term-missing --cov=yumroad. This run will create a file tests/cassettes/new_stripe_session.yaml which will record details about the request and the response.

Take a look at the file to make sure it matches what you expect. An example cassette is provided with the source code included in the book.

Annotate the other tests in test_product.py that make a network request.

@vcr.use_cassette('tests/cassettes/new_stripe_session.yaml',
                  filter_headers=['authorization'], record_mode='once')
def test_creation(client, init_database, authenticated_request):
    response = client.post(url_for('product.create'),
                            data=dict(name='test product', description='is persisted', price=5),
                            follow_redirects=True)

    assert response.status_code == 200
    assert b'test product' in response.data
    assert b'Buy for $5.00' in response.data

@vcr.use_cassette('tests/cassettes/new_stripe_session.yaml',
                  filter_headers=['authorization'], record_mode='once')
def test_post_checkout_success_page(client, init_database, user_with_product):
    product = Product.query.first()
    response = client.get(url_for('product.post_checkout', product_id=product.id,
                                  status='success', session_id='test_1'), follow_redirects=True)

    assert response.status_code == 200
    assert b'You will receive an email shortly' in response.data

@vcr.use_cassette('tests/cassettes/new_stripe_session.yaml',
                  filter_headers=['authorization'], record_mode='once')
def test_post_checkout_fail_page(client, init_database, user_with_product):
    product = Product.query.first()
    response = client.get(url_for('product.post_checkout', product_id=product.id,
                                  status='cancel', session_id='test_1'), follow_redirects=True)

    assert response.status_code == 200
    assert b'There was an error while attempting' in response.data

Now even if you go offline, these tests will still pass thanks to vcrpy.

In order to test our webhooks, we have to be able to mock up signed webhook requests. To do that, we'll need to ensure that the TestConfig in config.py is using a specific key and write a few functions to mock what Stripe is doing their own end.

In config.py:

class TestConfig(BaseConfig):
    ...
    STRIPE_WEBHOOK_KEY = 'whsec_test_secret'

Create a test suite called test_checkout.py in our tests folder.

We'll need a function to generate a signed header from arbitrary webhook data. This setup will help us get that.

import time
import json

from flask import url_for
import pytest
import stripe
import vcr

from yumroad.models import db, Product, Store, User
from yumroad.extensions import checkout

DUMMY_WEBHOOK_SECRET = 'whsec_test_secret'

def generate_header(payload, secret=DUMMY_WEBHOOK_SECRET, **kwargs):
    timestamp = kwargs.get("timestamp", int(time.time()))
    scheme = kwargs.get("scheme", stripe.WebhookSignature.EXPECTED_SCHEME)
    signature = kwargs.get("signature", None)
    if signature is None:
        payload_to_sign = "%d.%s" % (timestamp, payload)
        signature = stripe.WebhookSignature._compute_signature(
            payload_to_sign, secret
        )
    header = "t=%d,%s=%s" % (timestamp, scheme, signature)
    return {'Stripe-Signature': header}

def mock_webhook(event_name, data=None, webhook_secret=DUMMY_WEBHOOK_SECRET):
    payload = {}
    payload['type'] = event_name
    payload['data'] = {}
    payload['data']['object'] = data or {}
    data = json.dumps(payload)
    return data, generate_header(payload=data, secret=webhook_secret)

This lesson preview is part of the Fullstack Flask: Build a Complete SaaS App with Flask course and can be unlocked immediately with a single-time purchase. Already have access to this course? Log in here.

Unlock This Course

Get unlimited access to Fullstack Flask: Build a Complete SaaS App with Flask with a single-time purchase.

Thumbnail for the \newline course Fullstack Flask: Build a Complete SaaS App with Flask