(dsn, caplog)
| 319 | @pytest.mark.crdb_skip(class="st">"backend pid") |
| 320 | @pytest.mark.crdb_skip(class="st">"copy") |
| 321 | async def test_active_close(dsn, caplog): |
| 322 | caplog.set_level(logging.WARNING, logger=class="st">"psycopg.pool") |
| 323 | |
| 324 | async with pool.AsyncConnectionPool(dsn, min_size=1) as p: |
| 325 | conn = await p.getconn() |
| 326 | pid = conn.info.backend_pid |
| 327 | conn.pgconn.exec_(bclass="st">"copy (select * from generate_series(1, 10)) to stdout") |
| 328 | assert conn.info.transaction_status == TransactionStatus.ACTIVE |
| 329 | await p.putconn(conn) |
| 330 | |
| 331 | async with p.connection() as conn2: |
| 332 | assert conn2.info.backend_pid != pid |
| 333 | assert conn2.info.transaction_status == TransactionStatus.IDLE |
| 334 | |
| 335 | assert len(caplog.records) == 2 |
| 336 | assert class="st">"ACTIVE" in caplog.records[0].message |
| 337 | assert class="st">"BAD" in caplog.records[1].message |
| 338 | |
| 339 | |
| 340 | @pytest.mark.crdb_skip(class="st">"backend pid") |
nothing calls this directly
no test coverage detected