(
aconn_cls, aconn, apipeline, dsn, caplog
)
| 84 | |
| 85 | @pytest.mark.crdb_skip("pg_terminate_backend") |
| 86 | async def test_context_inerror_rollback_no_clobber( |
| 87 | aconn_cls, aconn, apipeline, dsn, caplog |
| 88 | ): |
| 89 | if apipeline: |
| 90 | # Only 'aconn' is possibly in pipeline mode, but the transaction and |
| 91 | # checks are on 'conn2'. |
| 92 | pytest.skip("not applicable") |
| 93 | caplog.set_level(logging.WARNING, logger="psycopg") |
| 94 | |
| 95 | with pytest.raises(ZeroDivisionError): |
| 96 | async with await aconn_cls.connect(dsn) as conn2: |
| 97 | async with conn2.transaction(): |
| 98 | await conn2.execute("select 1") |
| 99 | await aconn.execute( |
| 100 | "select pg_terminate_backend(%s::int)", |
| 101 | [conn2.pgconn.backend_pid], |
| 102 | ) |
| 103 | 1 / 0 |
| 104 | |
| 105 | assert len(caplog.records) == 1 |
| 106 | rec = caplog.records[0] |
| 107 | assert rec.levelno == logging.WARNING |
| 108 | assert "in rollback" in rec.message |
| 109 | |
| 110 | |
| 111 | @pytest.mark.crdb_skip("copy") |
nothing calls this directly
no test coverage detected
searching dependent graphs…