0

我正在使用 python 3.3、金字塔、sqlalchemy、psygopg2。我正在使用测试 postgres db 进行单元测试。我为鼻子运行设置了 101 个单元测试。在测试 101 中,我得到:

nose.proxy.OperationalError: (OperationalError) FATAL: 抱歉,客户已经太多了

从回溯看来,异常被抛出

......./venv/lib/python3.3/site-packages/SQLAlchemy-0.8.2-py3.3.egg/sqlalchemy/pool.py”,第 368 行,在 __connect

   connection = self.__pool._creator()

每次测试后可能没有运行 tearDown() ?Postgresql的连接池限制不是一次100吗?

这是我的 BaseTest 课程:

class BaseTest(object):
    def setup(self):
        self.request = testing.DummyRequest()
        self.config = testing.setUp(request=self.request)
        self.config.scan('../models')
        sqlalchemy_url = 'postgresql://<user>:<pass>@localhost:5432/<db>'
        engine = create_engine(sqlalchemy_url)
        DBSession = scoped_session(sessionmaker(extension=ZopeTransactionExtension()))
        DBSession.configure(bind=engine)
        Base.metadata.bind = engine
        Base.metadata.create_all(engine)
        self.dbsession = DBSession

    def tearDown(self):
        testing.teardown()

我的测试类继承自 BaseTest:

class TestUser(BaseTest):
    def __init__(self, dbsession = None):
        if dbsession:
            self.dbsession = dbsession
    
    def test_create_user(self):
        ......
        ......

其中一个测试类测试多对多关系,因此在该测试类中,我首先创建满足外键关系所需的记录:

from tests.test_user import TestUser
from tests.test_app import TestApp
class TestAppUser(BaseTest):
    def __init__(self, dbsession = None):
        if dbsession:
            self.dbsession = dbsession
    
    def create_app_user(self):
        test_app = TestApp(self.dbsession)
        test_user = TestUser(self.dbsession)
        test_app.request = testing.DummyRequest()
        test_user.request = testing.DummyRequest()
        app = test_app.create_app()
        user = test_user.create_user()
        ......

我将 dbsession 传递给 TestApp 和 TestUser 类...我认为这是问题的根源,但我不确定。

任何帮助是极大的赞赏。谢谢。

4

1 回答 1

1

Pyramid 与 SQLAlchemy 无关。Pyramid 的 API 中没有任何地方可以以 Pyramid 真正关心的方式链接任何 SQLAlchemy 配置。因此,金字塔对testing.tearDown()连接不做任何事情。怎么可能?它不知道它们的存在。

您正在将作用域会话与单元测试一起使用,这实际上没有多大意义,因为您的单元测试可能没有线程化。所以现在你正在创建线程本地会话而不是清理它们。它们不是垃圾收集的,因为它们是线程本地的。您也没有手动关闭这些连接,因此连接池认为它们仍在使用中。

您是否有理由ZopeTransactionExtension在测试中需要它?您是transaction在测试中使用该软件包,还是pyramid_tm?在测试中,如果您不知道某事做什么,那么它不应该存在。你是create_all()从你的setUp()方法调用的?这将是非常缓慢的自省数据库并在每个请求上创建表。哎哟。

class BaseTest(object):
    def setUp(self):
        self.request = testing.DummyRequest()
        self.config = testing.setUp(request=self.request)
        self.config.scan('../models')
        sqlalchemy_url = 'postgresql://<user>:<pass>@localhost:5432/<db>'
        self.engine = create_engine(sqlalchemy_url)
        Base.metadata.create_all(bind=self.engine)
        self.sessionmaker = sessionmaker(bind=self.engine)
        self.sessions = []

    def makeSession(self, autoclose=True):
        session = self.sessionmaker()
        if autoclose:
            self.sessions.append(session)

    def tearDown(self):
        for session in self.sessions:
            session.close()
        self.engine.dispose()
        testing.teardown()
于 2013-07-31T18:08:14.147 回答