Flask unittest and sqlalchemy using all connections

Question

I've just run into an issue running unittests on my flask app after I had roughly 100 unittests. All unittests will pass, but when run all at once they will fail with the following error:

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

Everything is running in a virtualbox/vagrant/ubuntu12.04 instance on local machine. My postgres max_connections is set to 100 so I'm assuming that the connections aren't closing and after running 100 tests I use up all the available ones.

This person Flask unit tests with SQLAlchemy and PostgreSQL exhausts db connections looks like they are having the same exact problem. Mike/Zzzeek (sqlalchemy dev) even responded to it saying that something may be happening in create_app() so I've included that as well below.

Does this mean I'm not closing my connections somewhere? All of these errors are triggered by db.create_all() in my setUp() method of my unittest.

# 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

Possible duplicate of Flask unit tests with SQLAlchemy and PostgreSQL exhausts db connections — Dag Høidahl
– Dag Høidahl, Commented Feb 4, 2016 at 10:36

Juan-Pablo Scaletti · Accepted Answer · 2015-01-19 21:18:59Z

UPDATE: Tested and fixed

Instead of making a new connection and re-creating your database every time (slow), you can use subsessions and do a rollback after each test.

The connection are reused, so this also fix the problem you're having.

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()

If you need to also make an instance of the application for each test, just add it to the setUp method but leave it also in setUpClass.

Full test example below requires flask_sqlalchemy and psycopg2. Create a test database named "test" and set its connection limit to 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()

setUpClass is run before create_app() is called so there is no db initialized yet. Although this is interesting for unittest performance. Even if it did work I think I would still have some sort of issue in my code not closing connections. This would just be a workaround to a problem that might get exposed in production.
Juan-Pablo, Thank you very much for taking the time to answer. I don't care about points at all but I answered my own question, so the next person wouldn't be as frustrated. I believe your solution is maybe an overall better solution to the problem as it solves it and clearly makes the tests run waaaay faster, but I was unable to get it to work either. I could run what you have but, I was violating unique constraints on my database since the database wasn't really rolling back, between tests. I will continue to research. Thank you for your time
This has the issue that you cant run tests in parallel anymore, since it is required to run them sequentially since the transactions cant overlap.

Community · Accepted Answer · 2017-05-23 11:58:41Z

8

I found the answer here -- https://stackoverflow.com/a/17998485/1870623 and a great explination here -- https://stackoverflow.com/a/16390645/1870623

The solution is to add db.get_engine(self.app).dispose() to the 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

edited May 23, 2017 at 11:58

CommunityBot

11 silver badge

answered Aug 21, 2013 at 19:06

Adam P

3312 silver badges11 bronze badges

Collectives™ on Stack Overflow

Flask unittest and sqlalchemy using all connections

2 Answers 2

4 Comments

Comments

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

4 Comments

Comments

Linked

Related