4

約100回のユニットテストを行った後、フラスコアプリでユニットテストを実行する際に問題が発生しました。すべての単体テストはパスしますが、一度にすべて実行すると、次のエラーで失敗します。

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

すべてがローカル マシンの virtualbox/vagrant/ubuntu12.04 インスタンスで実行されています。私のpostgres max_connectionsは100に設定されているので、接続が閉じられていないと仮定しています.100回のテストを実行した後、利用可能なものをすべて使い果たします。

この人物は、SQLAlchemy と PostgreSQL を使用した Flask 単体テストで db 接続を使い果たし、まったく同じ問題を抱えているように見えます。Mike/Zzzeek (sqlalchemy dev) は、create_app() で何かが起こっている可能性があると答えたので、それも以下に含めました。

これは、どこかで接続を閉じていないということですか? これらのエラーはすべてdb.create_all()、ユニットテストの setUp() メソッドでトリガーされます。

# 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
4

2 に答える 2

13

更新：テストおよび修正済み

毎回新しい接続を作成してデータベースを再作成する (遅い) 代わりに、サブセッションを使用して、各テストの後にロールバックを実行できます。

接続が再利用されるため、これにより問題も解決されます。

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()
于 2013-08-17T20:42:38.467 に答える
7

私はここで答えを見つけました - https://stackoverflow.com/a/17998485/1870623ここで素晴らしい説明 - https://stackoverflow.com/a/16390645/1870623

db.get_engine(self.app).dispose()解決策は、tearDown()に追加することです

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
于 2013-08-21T19:06:12.933 に答える