def _check_open_getconn(self) -> None:
if self._closed:
if self._opened:
- raise PoolClosed(
- f"the pool {self.name!r} has already been closed"
- )
+ raise PoolClosed(f"the pool {self.name!r} is already closed")
else:
- raise PoolClosed(
- f"the pool {self.name!r} has not been opened yet"
- )
+ raise PoolClosed(f"the pool {self.name!r} is not open yet")
def get_stats(self) -> Dict[str, int]:
"""
def test_open_explicit(dsn):
p = pool.ConnectionPool(dsn, open=False)
assert p.closed
- with pytest.raises(pool.PoolClosed, match="has not been opened yet"):
+ with pytest.raises(pool.PoolClosed, match="is not open yet"):
p.getconn()
with pytest.raises(pool.PoolClosed):
finally:
p.close()
- with pytest.raises(pool.PoolClosed, match="has already been closed"):
+ with pytest.raises(pool.PoolClosed, match="is already closed"):
p.getconn()
with pytest.raises(pool.PoolClosed):
await p.getconn()
- with pytest.raises(pool.PoolClosed, match="has not been opened yet"):
+ with pytest.raises(pool.PoolClosed, match="is not open yet"):
async with p.connection():
pass
finally:
await p.close()
- with pytest.raises(pool.PoolClosed, match="has already been closed"):
+ with pytest.raises(pool.PoolClosed, match="is already closed"):
await p.getconn()