[PYTHON] Rollback DB pour chaque test avec Flask + SQLAlchemy

Pour tester les applications Flask, vous pouvez utiliser la base de données mémoire de sqlite pour les applications simples, mais pour les applications complexes, vous souhaitez souvent exécuter en utilisant le même SGBDR que votre environnement de production.

À ce moment-là, il est lent d'initialiser la base de données pour chaque test, je voudrais donc le gérer par rollback à chaque fois sans engagement, mais dans un test qui fait plusieurs requêtes HTTP à l'aide de Flask.test_client ou WebTest, il chevauche la requête. Vous devez reprendre les données.

Pour y parvenir,

1. Remplacez session.commit () par session.flush (), session.expire_all () pendant le test
2. À la fin de la requête, généralement session.remove (), et pendant le test session.expire_all ()
3. Faites session.remove () pour chaque test.

Je personnalise.

session.flush () écrit toutes les modifications gérées par la session (Unité de travail) dans la base de données. session.expire_all () expire tous les objets gérés par la session et les récupère de la base de données la prochaine fois qu'ils les utilisent. Vous pouvez maintenant exécuter le test avec les valeurs lues à partir du DB, pas les valeurs en mémoire. En particulier, en faisant expire_all () à la fin de la requête testée, s'il y a une omission commit (), la valeur sera correctement effacée dans la requête suivante. session.remove () annule la transaction pour cette session et renvoie la connexion au pool de connexions.

Je publierai la méthode de personnalisation.

sessionmanager.py

from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session, sessionmaker, Session


class TestingSession(Session):
    """Session for testing."""

    def commit(self):
        u"""commit()Flusn(), expire_all()Émuler avec."""
        self.flush()
        self.expire_all()


class SessionManager(object):

    def __init__(self, app=None):
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        self._create_session(app)
        #Lors des tests, le côté test gère la vie de la session.
        if not app.testing:
            app.teardown_appcontext(self.exit_sessions)

    def _create_session(self, app, testing=False):
        self.session = scoped_session(sessionmaker(
            bind=create_engine(app.config['DATABASE_DSL']),
            class_=TestingSession if testing else Session,
            expire_on_commit=False))

    def _exit_session(self, response_or_exc):
        self.session.remove()
        return response_or_exc

myapp.py

import flask
from . import sessionmanager

db = sessionmanager.SessionManager()

def get_app(testing=None):
    app = flask.Flask(__name__)

    app.config.from_envvar('MYAPP_SETTING')
    if testing is not None:
        app.testing = testing

    # db.init_app()Est l'application.Après avoir réglé le test.
    db.init_app(app)

    #Enregistrez la vue ici

    return app

test_app.py

import unittest
import myapp

class AppTestCase(unittest.TestCase):
    def setUp(self):
        self.app = myapp.get_app(testing=True)

        #Restauration pour chaque test
        self.addCleanup(myapp.db.session.remove)

        @self.app.after_request:
        def after_request(response):
            u"""Oubliez les modifications non engagées à chaque demande"""
            myapp.db.session.expire_all()
            return response

Flask-SQLAlchemy ne peut pas être pris en charge tel quel car il gère les sessions indépendamment, mais je pense que la même chose peut être faite en surchargeant Session.remove () et en la personnalisant pour qu'elle n'exécute normalement que expire_all ().

from flask.ext.sqlalchemy import SQLAlchemy as BaseSQLAlchemy, SignallingSession

class TestSession(SignallingSession):
    def commit(self):
        self.flush()
        self.expire_all()

    def remove(self):
        self.expire_all()

    def real_remove(self):
        super(TestSession, self).remove()


class SQLAlchemy(BaseSQLAlchemy):
    def create_session(self, options):
        if self.app.testing:
            return TestSession(**options)
        else:
            return SignallingSession(**options)