Function TestPipelineGetResultsHandlesPartiallyReadResults

pgconn/pgconn_test.go:3965–4004

(t *testing.T)

}

func TestPipelineGetResultsHandlesPartiallyReadResults(t *testing.T) {
	t.Parallel()

	ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
	defer cancel()

	pgConn, err := pgconn.Connect(ctx, os.Getenv("PGX_TEST_DATABASE"))
	require.NoError(t, err)
	defer closeConn(t, pgConn)

	sd, err := pgConn.Prepare(ctx, "ps", "select n from generate_series($1::int, $2::int) n", nil)
	require.NoError(t, err)

	pipeline := pgConn.StartPipeline(ctx)
	pipeline.SendQueryStatement(sd, [][]byte{[]byte("1"), []byte("3")}, nil, nil)
	pipeline.SendQueryStatement(sd, [][]byte{[]byte("5"), []byte("7")}, nil, nil)
	err = pipeline.Sync()
	require.NoError(t, err)

	results, err := pipeline.GetResults()
	require.NoError(t, err)
	rr, ok := results.(*pgconn.ResultReader)
	require.Truef(t, ok, "expected ResultReader, got: %#v", results)
	require.True(t, rr.NextRow())
	require.Equal(t, "1", string(rr.Values()[0]))

	results, err = pipeline.GetResults()
	require.NoError(t, err)
	rr, ok = results.(*pgconn.ResultReader)
	require.Truef(t, ok, "expected ResultReader, got: %#v", results)
	require.True(t, rr.NextRow())
	require.Equal(t, "5", string(rr.Values()[0]))
	require.True(t, rr.NextRow())
	require.Equal(t, "6", string(rr.Values()[0]))

	err = pipeline.Close()
	require.NoError(t, err)

	ensureConnValid(t, pgConn)
}

func TestPipelineCloseReadsUnreadResults(t *testing.T) {
	t.Parallel()