テストスイート内の外部トランザクションの使用

Question

External transactions

モデルにモデルをコミットし、各テスト後にロールバックする必要があります。純粋なsqlalchemy実装（最下部）は、私が期待している通りに動作します。ただし、flask-sqlalchemyの実装では、各テスト後にコミットされたモデルをロールバックできません。

flask-sqlalchemyのスコープ付きセッションをエンジンの接続にバインドする方法はありますか。

ジェネリックflaskアプリ：

from flask import Flask 
from flask_sqlalchemy import SQLAlchemy 


app = Flask(__name__) 
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite://' 
db = SQLAlchemy(app) 


class A(db.Model): 
    id = db.Column(db.Integer, primary_key=True) 

flask-sqlalchemyは、テストケース（ロールバックされません）：

from app import * 
from unittest import TestCase 


class Test(TestCase): 

    def setUp(self): 
     self.connection = db.engine.connect() 
     self.transaction = self.connection.begin() 

     options = dict(bind=self.connection, binds={}) 
     self.session = db.create_scoped_session(options=options) 
     db.session = self.session 

     self.addCleanup(self.cleanup) 

    def cleanup(self): 
     self.transaction.rollback() 
     self.connection.close() 
     self.session.remove() 

    @classmethod 
    def setUpClass(cls): 
     db.create_all() 

    @classmethod 
    def tearDownClass(cls): 
     pass 

    def test_1(self): 
     a = A() 
     db.session.add(a) 
     db.session.commit() 

     assert len(db.session.query(A).all()) == 1 

    def test_2(self): 
     assert len(db.session.query(A).all()) == 0 # len is 1 

sqlalchemyテストケース（作品）：

from sqlalchemy import * 
from sqlalchemy.orm import * 
from sqlalchemy.ext.declarative import declarative_base 
from unittest import TestCase 


Base = declarative_base() 


class A(Base): 
    __tablename__ = 'test' 
    id = Column(Integer, primary_key=True) 


class Test(TestCase): 

    def setUp(self): 
     self.connection = e.connect() 
     self.transaction = self.connection.begin() 

     # Begin scoped session 
     factory = sessionmaker(bind=self.connection) 
     self.session = scoped_session(factory) 

     self.addCleanup(self.cleanup) 

    def cleanup(self): 
     self.session.close() 
     self.transaction.rollback() 
     self.connection.close() 

    @classmethod 
    def setUpClass(cls): 
     global e 

     e = create_engine("sqlite://") 
     Base.metadata.create_all(e) 

    @classmethod 
    def tearDownClass(cls): 
     pass 

    def test_1(self): 
     a = A() 
     self.session.add(a) 
     self.session.commit() 

     assert len(self.session.query(A).all()) == 1 

    def test_2(self): 
     assert len(self.session.query(A).all()) == 0 

Answer 1

私は同じ問題を抱えていましたおそらく私の修正はあなたのものと同じです。

これは私の確立とティアダウンは今です：

import unittest 
from my_app import db 

class Test(unittest.TestCase): 
    def setUp(self): 
     self.connection = db.engine.connect() 
     self.trans = self.connection.begin() 
     db.session.configure(bind=self.connection, binds={}) 

    def tearDown(self): 
     self.trans.rollback() 
     self.connection.close() 
     db.session.remove() 

実際に追加された私のためにそれをやった：これはそれを修正するが、それがなかった理由を私は実際に調査していない

binds={} 

私のために。誰かがなぜバインドする空の言葉が問題を解決するのか分かっているのを聞くのが大好きです。