0

私はpython 3.3、ピラミッド、sqlalchemy、psygopg2を使用しています。単体テストにテスト postgres db を使用しています。私は鼻を走らせるために101個のユニットテストをセットアップしました。テスト101で私は得る:

nodes.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):
        ......
        ......

テスト クラスの 1 つは多対多の関係をテストするため、そのテスト クラスでは、最初に外部キー関係を満たすために必要なレコードを作成します。

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 設定をリンクする場所はありません。したがって、Pyramidtesting.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 に答える