燒瓶unittest和sqlalchemy使用所有連接

Question

我剛剛遇到一個問題，在我的燒瓶應用上運行unittests後，我有大約100個單元測試。所有的單元測試會通過，但是當他們立刻將出現以下錯誤失敗運行所有：

OperationalError: (OperationalError) FATAL: remaining connection slots are reserved for non-replication superuser connections 

一切都在本地機器上的VirtualBox /流浪者/ ubuntu12.04實例中運行。我的postgres max_connections設置爲100，所以我假設連接不關閉，運行100次測試後，我用完了所有可用的連接。

此人Flask unit tests with SQLAlchemy and PostgreSQL exhausts db connections看起來像他們有相同的確切問題。 Mike/Zzzeek（sqlalchemy dev）甚至迴應說，可能在create_app（）中發生了一些事情，所以我在下面也包含了這一點。

這是否意味着我沒有關閉我的連接？所有這些錯誤都是由我的unittest的setUp（）方法中的db.create_all()觸發的。

＃test.py

class TestCase(DataMixin, Base): 
    """Base test class""" 

    def create_app(self): 
     return create_app(TestConfig()) 

    def setUp(self): 
     db.create_all() 

    def tearDown(self): 
     db.session.remove() 
     db.drop_all() 

＃app.py

def create_app(config=None): 
    app = Flask(__name__) 

    # Config 
    app.config.from_object(BaseConfig()) 
    if config is not None: 
     app.config.from_object(config) 

    # Extensions 
    db.init_app(app) 
    mail.init_app(app) 
    bcrypt.init_app(app) 

    # Blueprints 
    app.register_blueprint(core_blueprint, url_prefix='/') 
    app.register_blueprint(accounts_blueprint, url_prefix='/account') 
    app.register_blueprint(admin_blueprint, url_prefix='/admin') 
    app.register_blueprint(cart_blueprint, url_prefix='/cart') 

    # Login Manager 
    login_manager.setup_app(app, add_context_processor=True) 
    login_manager.login_view = "accounts.login" 
    login_manager.user_callback = load_user 

    # Templates 
    app.jinja_env.globals['is_admin'] = is_admin 
    app.jinja_env.globals['is_staff'] = is_staff 

    @app.context_processor 
    def inject_cart(): 
     cart = count = None 
     if current_user.is_authenticated(): 
      cart = current_user.get_cart() 
     return dict(cart=cart) 

    # Error Handling 
    @app.errorhandler(404) 
    def page_not_found(error): 
     return render_template('404.html'), 404 

    return app 

Answer 1

我找到了答案在這裏 - https://stackoverflow.com/a/17998485/1870623和一個偉大的explination在這裏 - https://stackoverflow.com/a/16390645/1870623

的解決方案是增加db.get_engine(self.app).dispose()到拆機（）

class TestCase(Base): 
    def setUp(self): 
     db.create_all() 

    def tearDown(self): 
     db.session.remove() 
     db.drop_all() 
     db.get_engine(self.app).dispose() # This 

Answer 2

UPDATE：測試和固定

而不是建立新連接，每重新創建數據庫時間（慢），您可以使用子會話並在每次測試後進行回滾。

連接被重用，所以這也解決了你遇到的問題。

class TestCase(Base): 

    @classmethod 
    def setUpClass(cls): 
     cls.app = create_app(MyConfig()) 
     cls.client = cls.app.test_client() 
     cls._ctx = cls.app.test_request_context() 
     cls._ctx.push() 
     db.create_all() 

    @classmethod 
    def tearDownClass(cls): 
     db.session.remove() 
     db.drop_all() 
     db.get_engine(cls.app).dispose() 

    def setUp(self): 
     self._ctx = self.app.test_request_context() 
     self._ctx.push() 
     db.session.begin(subtransactions=True) 

    def tearDown(self): 
     db.session.rollback() 
     db.session.close() 
     self._ctx.pop() 

如果您還需要爲每個測試的應用程序的實例，只是將它添加到setUp方法，但離開它也setUpClass。

下面的完整測試示例需要flask_sqlalchemy和psycopg2。創建一個名爲「測試」的測試數據庫和它的連接限制設置爲15

from flask import Flask 
from flask.ext.sqlalchemy import SQLAlchemy 
from unittest import TestCase as Base 


db = SQLAlchemy() 

def create_app(config=None): 
    app = Flask(__name__) 
    app.config.from_object(config) 
    db.init_app(app) 
    return app 


class MyConfig(object): 
    SQLALCHEMY_DATABASE_URI = "postgresql://localhost/test" 
    TESTING = True 


class TestCase(Base): 
    @classmethod 
    def setUpClass(cls): 
     cls.app = create_app(MyConfig()) 
     cls.client = cls.app.test_client() 
     cls._ctx = cls.app.test_request_context() 
     cls._ctx.push() 
     db.create_all() 

    @classmethod 
    def tearDownClass(cls): 
     db.session.remove() 
     db.drop_all() 

    def setUp(self): 
     self._ctx = self.app.test_request_context() 
     self._ctx.push() 
     db.session.begin(subtransactions=True) 

    def tearDown(self): 
     db.session.rollback() 
     db.session.close() 
     self._ctx.pop() 


class TestModel(TestCase): 

    def test_01(self): 
     pass 

    def test_02(self): 
     pass 

    def test_03(self): 
     pass 

    def test_04(self): 
     pass 

    def test_05(self): 
     pass 

    def test_06(self): 
     pass 

    def test_07(self): 
     pass 

    def test_08(self): 
     pass 

    def test_09(self): 
     pass 

    def test_10(self): 
     pass 

    def test_11(self): 
     pass 

    def test_12(self): 
     pass 

    def test_13(self): 
     pass 

    def test_14(self): 
     pass 

    def test_15(self): 
     pass 

    def test_16(self): 
     pass 


if __name__ == "__main__": 
    import unittest 
    unittest.main()