[PYTHON] Rollback DB für jeden Test mit Flask + SQLAlchemy

Zum Testen von Flask-Apps können Sie die Speicher-DB von sqlite für einfache Apps verwenden. Für komplexe Apps möchten Sie jedoch häufig dasselbe RDBMS wie Ihre Produktionsumgebung verwenden.

Zu diesem Zeitpunkt ist die Initialisierung der Datenbank für jeden Test langsam, daher möchte ich sie jedes Mal per Rollback ohne Festschreiben behandeln. Bei einem Test, bei dem mehrere HTTP-Anforderungen mit Flask.test_client oder WebTest ausgeführt werden, wird die Anforderung jedoch überspannt. Sie müssen die Daten übernehmen.

Um das zu erreichen,

1. Ersetzen Sie session.commit () während des Tests durch session.flush (), session.expire_all ()
2. Am Ende der Anforderung normalerweise session.remove () und während des Testens session.expire_all ()
3. Führen Sie für jeden Test session.remove () aus.

Ich passe an.

session.flush () schreibt alle von der Sitzung (Arbeitseinheit) verwalteten Änderungen in die Datenbank. session.expire_all () lässt alle von der Sitzung verwalteten Objekte ablaufen und ruft sie bei der nächsten Verwendung aus der Datenbank ab. Jetzt können Sie den Test mit den aus der Datenbank gelesenen Werten ausführen, nicht mit den Werten im Speicher. Insbesondere durch Ausführen von expire_all () am Ende der zu testenden Anforderung wird der Wert bei der nächsten Anforderung ordnungsgemäß gelöscht, wenn eine Commit () - Auslassung vorliegt. session.remove () setzt die Transaktion für diese Sitzung zurück und gibt die Verbindung zum Verbindungspool zurück.

Ich werde die Anpassungsmethode veröffentlichen.

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()Emulieren mit."""
        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)
        #Während des Tests verwaltet die Testseite die Lebensdauer der Sitzung.
        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()Ist App.Nach dem Einstellen des Tests.
    db.init_app(app)

    #Registrieren Sie die Ansicht hier

    return app

test_app.py

import unittest
import myapp

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

        #Rollback für jeden Test
        self.addCleanup(myapp.db.session.remove)

        @self.app.after_request:
        def after_request(response):
            u"""Vergessen Sie nicht festgeschriebene Änderungen bei jeder Anfrage"""
            myapp.db.session.expire_all()
            return response

Flask-SQLAlchemy kann nicht unterstützt werden, da es Sitzungen unabhängig verwaltet. Ich denke jedoch, dass dies auch durch Überschreiben von Session.remove () und Anpassen so geschehen kann, dass normalerweise nur expire_all () ausgeführt wird.

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)