(dsn, caplog)
| 232 | |
| 233 | @pytest.mark.crdb_skip("backend pid") |
| 234 | async def test_reset_badstate(dsn, caplog): |
| 235 | caplog.set_level(logging.WARNING, logger="psycopg.pool") |
| 236 | |
| 237 | async def reset(conn): |
| 238 | await conn.execute("reset all") |
| 239 | |
| 240 | async with pool.AsyncConnectionPool(dsn, min_size=1, reset=reset) as p: |
| 241 | async with p.connection() as conn: |
| 242 | await conn.execute("select 1") |
| 243 | pid1 = conn.info.backend_pid |
| 244 | |
| 245 | async with p.connection() as conn: |
| 246 | await conn.execute("select 1") |
| 247 | pid2 = conn.info.backend_pid |
| 248 | |
| 249 | assert pid1 != pid2 |
| 250 | assert caplog.records |
| 251 | assert "INTRANS" in caplog.records[0].message |
| 252 | |
| 253 | |
| 254 | @pytest.mark.crdb_skip("backend pid") |
nothing calls this directly
no test coverage detected