| 115 | eq_(row, expected.pop(0)) |
| 116 | |
| 117 | def test_no_connect_on_recreate(self): |
| 118 | def creator(): |
| 119 | raise Exception("no creates allowed") |
| 120 | |
| 121 | for cls in ( |
| 122 | pool.SingletonThreadPool, |
| 123 | pool.StaticPool, |
| 124 | pool.QueuePool, |
| 125 | pool.NullPool, |
| 126 | pool.AssertionPool, |
| 127 | ): |
| 128 | p = cls(creator=creator) |
| 129 | p.dispose() |
| 130 | p2 = p.recreate() |
| 131 | assert p2.__class__ is cls |
| 132 | |
| 133 | mock_dbapi = MockDBAPI() |
| 134 | p = cls(creator=mock_dbapi.connect) |
| 135 | conn = p.connect() |
| 136 | conn.close() |
| 137 | mock_dbapi.connect.side_effect = Exception("error!") |
| 138 | p.dispose() |
| 139 | p.recreate() |
| 140 | |
| 141 | def test_info(self): |
| 142 | p = self._queuepool_fixture(pool_size=1, max_overflow=0) |