(dsn, caplog)
| 253 | |
| 254 | @pytest.mark.crdb_skip("backend pid") |
| 255 | async def test_reset_broken(dsn, caplog): |
| 256 | caplog.set_level(logging.WARNING, logger="psycopg.pool") |
| 257 | |
| 258 | async def reset(conn): |
| 259 | async with conn.transaction(): |
| 260 | await conn.execute("WAT") |
| 261 | |
| 262 | async with pool.AsyncConnectionPool(dsn, min_size=1, reset=reset) as p: |
| 263 | async with p.connection() as conn: |
| 264 | await conn.execute("select 1") |
| 265 | pid1 = conn.info.backend_pid |
| 266 | |
| 267 | async with p.connection() as conn: |
| 268 | await conn.execute("select 1") |
| 269 | pid2 = conn.info.backend_pid |
| 270 | |
| 271 | assert pid1 != pid2 |
| 272 | assert caplog.records |
| 273 | assert "WAT" in caplog.records[0].message |
| 274 | |
| 275 | |
| 276 | @pytest.mark.crdb_skip("backend pid") |
nothing calls this directly
no test coverage detected