postgresql – 使用所有连接的Flask unittest和sqlalchemy

在我进行了大约100次单元测试后,我刚刚遇到了在我的烧瓶应用程序上运行unittests的问题.所有单元测试都将通过,但是当一次运行时,它们将失败并出现以下错误：

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

一切都在本地计算机上的virtualBox / vagrant / 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

解决方法

更新：经过测试和修复

您不必每次(慢)创建新连接并重新创建数据库,而是可以使用子会话并在每次测试后执行回滚.

连接被重用,所以这也解决了你遇到的问题.

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.创建名为“test”的测试数据库,并将其连接限制设置为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()