(pgconn)
| 69 | |
| 70 | @pytest.fixture |
| 71 | def pipeline(pgconn): |
| 72 | nb, pgconn.nonblocking = pgconn.nonblocking, True |
| 73 | assert pgconn.nonblocking |
| 74 | pgconn.enter_pipeline_mode() |
| 75 | yield |
| 76 | if pgconn.pipeline_status: |
| 77 | pgconn.exit_pipeline_mode() |
| 78 | pgconn.nonblocking = nb |
| 79 | |
| 80 | |
| 81 | def _run_pipeline_communicate(pgconn, generators, commands, expected_statuses): |
nothing calls this directly
no test coverage detected