A 'SQLAlchemy' instance has already been registered when running more than one test function with pytest in flask application

Question

Pytest runs fine with one test function in its file but fails when a second function is introduced.

Tests pass individually or as one function but can't run two test functions in one class.

The error is

RuntimeError: A 'SQLAlchemy' instance has already been registered on this Flask app. Import and use that instance instead.

This uses the Miguel Grinberg flask application structure

TestClass Here's the test class. It can work with only one function and both assertions but not with two separate functions.

import json
import pytest
from app import create_app


@pytest.fixture
def app():
    app = create_app('testing')
    yield app


@pytest.fixture
def client(app):
    return app.test_client()


def test_oauth_redirect_with_valid_code(client, monkeypatch):
    with open("tests/membership/permanent_access_code_success.json") as json_file:
        success_data = json.load(json_file)
    monkeypatch.setattr("app.membership.controllers.exchange_temporary_code_for_permanent",lambda data: success_data )
    monkeypatch.setattr("app.membership.controllers.register_member", lambda data: (None, 200))
    response = client.get("/member/register?code=valid_code")
    assert response.status_code == 200


def test_oauth_redirect_with_invalid_code(client, monkeypatch):
    with open("tests/membership/permanent_access_code_fail.json") as json_file:
        fail_data = json.load(json_file)
    monkeypatch.setattr("app.membership.controllers.exchange_temporary_code_for_permanent",
                        lambda data: (fail_data, 200))
    monkeypatch.setattr("app.membership.controllers.register_member", lambda data: (None, 500))
    fail_response = client.get("/member/register?code=invalid_code")
    assert fail_response.status_code == 500

**Class Under Test - An excerpt of the relevant class **

from .models import Member, MembershipType
from threading import Thread
from app import db, app, scheduler
import datetime
import json
import urllib.parse
import urllib.request
from urllib.error import HTTPError
from sqlalchemy.exc import IntegrityError
from flask import request
from flask import Blueprint, current_app, render_template
import logging
from app import db, app
import os

from ..slack.slack_service import respond_to_slash_command, get_team_access_token, send_slack_message

membership_bp = Blueprint('membership_bp', __name__)


@membership_bp.route("/member/register", methods=["GET"])
def oauth_redirect():
    logger = current_app.logger
    code_param = request.args.get("code")
    logger.debug("Code param:")
    logger.debug(code_param)
    permanent_access_code_response = exchange_temporary_code_for_permanent(code_param)
    logger.debug("permanent_access_code_response full")
    logger.debug(permanent_access_code_response)
    if permanent_access_code_response[0]["ok"]:
        registered_member_details = permanent_access_code_response[0]
        logger.debug("registered_member_details response")
        logger.debug(registered_member_details)
        registration_result = register_member(registered_member_details)
        if registration_result[1] == 200:
            Thread(target=send_sales_message, args=(
                logger, registered_member_details["authed_user"]["id"],
                registered_member_details["team"]["id"], MembershipType.TRIAL)).start()
            return render_template('installed.html')
        else:
            return render_template('500.html', error_string=registration_result[0]), 500
    else:
        return render_template('500.html', error_string="Failed Slack authorisation, please retry clicking the "
                                                        "Install button below"), 500


def exchange_temporary_code_for_permanent(code):
    client_id = os.environ.get("SLACK_CLIENT_ID")
    client_secret = os.environ.get("SLACK_CLIENT_SECRET")
    data_form = {'code': code, "client_id": client_id, "client_secret": client_secret}
    current_app.logger.debug("Sending the following for code exchange")
    current_app.logger.debug(data_form)
    data_bytes = urllib.parse.urlencode(data_form).encode('utf-8')
    headers = {'Content-Type': 'application/x-www-form-urlencoded'}
    oauth_url = "https://slack.com/api/oauth.v2.access"
    req = urllib.request.Request(oauth_url, data_bytes, headers)
    try:
        with urllib.request.urlopen(req) as urllib_handler:
            response = urllib_handler.read().decode('utf-8')
            current_app.logger.debug("response received when exchanging code")
            current_app.logger.debug(response)
            return [json.loads(response), 200]

    except HTTPError as e:
        error_content = e.read()
        current_app.logger.error(error_content)
        return [error_content, 500]


def register_member(access_token_response):
    current_app.logger.debug("Register member to DB")
    try:
        authed_user = access_token_response["authed_user"]["id"]
    except KeyError as e:
        return [
            "You have attempted to refresh the install page, please click the install button on the home page instead",
            500]
    authed_user = access_token_response["authed_user"]["id"]
    scope = access_token_response["scope"]
    access_token = access_token_response["access_token"]
    team_id = access_token_response["team"]["id"]
    team_name = access_token_response["team"]["name"]
    enterprise = access_token_response["enterprise"]
    is_enterprise_install = access_token_response["is_enterprise_install"]
    membership_type = MembershipType.TRIAL
    expiration_date = datetime.datetime.now() + datetime.timedelta(days=7)
    new_member = Member(authed_user, scope, access_token, team_id, team_name, enterprise,
                        is_enterprise_install, membership_type, expiration_date)
    current_app.logger.debug(new_member)
    try:
        db.session.add(new_member)
        db.session.commit()
        return ["", 200]
    except IntegrityError as e:
        current_app.logger.error(e)
        return [
            "This Slack workspace already has an account, please email [email protected] to get it removed "
            "for a reinstall.",
            500]
    except Exception as e:
        current_app.logger.error(e)
        return ["Internal Server Error, please contact [email protected] for support", 500]

config.py This is primarily taken from Miguel Grinberg\s tutorial

import os
BASE_DIR = os.path.abspath(os.path.dirname(__file__))  


class Config:
    THREADS_PER_PAGE = 2
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    CSRF_ENABLED = False


    
    @staticmethod
    def init_app(app):
        pass

class DevelopmentConfig(Config):
    DB_PASSWORD = os.environ['DB_PASSWORD']
    SQLALCHEMY_DATABASE_URI = "..."

    DEBUG = True

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///' + os.path.join(BASE_DIR, 'data.sqlite')

class ProductionConfig(Config):
    DB_PASSWORD = os.environ['DB_PASSWORD']
    SQLALCHEMY_DATABASE_URI = "..."

    @classmethod
    def init_app(cls, app):
        Config.init_app(app)


config = {
    'development': DevelopmentConfig,
    'testing': TestingConfig,
    'production': ProductionConfig,

    'default': DevelopmentConfig
}

init.py and again it is primarily taken from Miguel Grinbergs flask tutorial

# Import the database object (db) from the main application module
# We will define this inside /app/__init__.py in the next sections.
# Import SQLAlchemy
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from config import config
import logging
import os
from flask_apscheduler import APScheduler
import socket


db = SQLAlchemy()
scheduler = APScheduler()

app = Flask(__name__)

def create_app(config_name):

    app.config.from_object(config[config_name])

    gunicorn_logger = logging.getLogger('gunicorn.error')
    gunicorn_logger.setLevel(logging.DEBUG)
    app.logger.handlers = gunicorn_logger.handlers
    app.logger.setLevel(gunicorn_logger.level)

    db.init_app(app)
    
   # Fix to ensure only one Gunicorn worker grabs the scheduled task
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(("127.0.0.1", 47200))
    except socket.error:
        pass
    else:
        scheduler.init_app(app)
        scheduler.start()

    from app.responses.controllers import responses_bp
    app.register_blueprint(responses_bp)

    from app.membership.controllers import membership_bp
    app.register_blueprint(membership_bp)

    return app
    

Answer 1

Seems fixtures are being run before every test. I ended up moving the fixtures code to a conftest.py file at the base of my tests directory and changing the decorator to say the fixture is per session

import pytest
from app import create_app


@pytest.fixture(scope="session", autouse=True)
def app():
    app = create_app('testing')
    yield app


@pytest.fixture
def client(app):
    return app.test_client()