Function test_flush_request

tests/pq/test_pipeline.py:77–85

(pgconn)

@pytest.mark.libpq(">= 14")
def test_flush_request(pgconn):
    assert pgconn.pipeline_status == pq.PipelineStatus.OFF
    pgconn.enter_pipeline_mode()
    pgconn.send_query_params(b"select $1", [b"1"], param_types=[25])
    pgconn.send_flush_request()
    r = pgconn.get_result()
    assert r.status == pq.ExecStatus.TUPLES_OK
    assert r.get_value(0, 0) == b"1"
    pgconn.exit_pipeline_mode()


@pytest.fixture