(pool_cls, aconn_cls, dsn, autocommit)
| 606 | @pytest.mark.crdb_skip("pg_terminate_backend") |
| 607 | @pytest.mark.parametrize("autocommit", [True, False]) |
| 608 | async def test_check_connection(pool_cls, aconn_cls, dsn, autocommit): |
| 609 | conn = await aconn_cls.connect(dsn) |
| 610 | await set_autocommit(conn, autocommit) |
| 611 | await pool_cls.check_connection(conn) |
| 612 | assert not conn.closed |
| 613 | assert conn.info.transaction_status == psycopg.pq.TransactionStatus.IDLE |
| 614 | |
| 615 | async with await aconn_cls.connect(dsn) as conn2: |
| 616 | await conn2.execute("select pg_terminate_backend(%s)", [conn.info.backend_pid]) |
| 617 | |
| 618 | with pytest.raises(psycopg.OperationalError): |
| 619 | await pool_cls.check_connection(conn) |
| 620 | |
| 621 | assert conn.closed |
| 622 | |
| 623 | |
| 624 | async def test_check_init(pool_cls, dsn): |
nothing calls this directly
no test coverage detected