Function TestConnCopyFromCanceled

pgconn/pgconn_test.go:2438–2483

(t *testing.T)

}

func TestConnCopyFromCanceled(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)

	_, err = pgConn.Exec(ctx, `create temporary table foo(
		a int4,
		b varchar
	)`).ReadAll()
	require.NoError(t, err)

	r, w := io.Pipe()
	go func() {
		for i := range 1_000_000 {
			a := strconv.Itoa(i)
			b := "foo " + a + " bar"
			_, err := w.Write(fmt.Appendf(nil, "%s,\"%s\"\n", a, b))
			if err != nil {
				return
			}
			time.Sleep(time.Microsecond)
		}
	}()

	ctx, cancel = context.WithTimeout(ctx, 100*time.Millisecond)
	copySql := "COPY foo FROM STDIN WITH (FORMAT csv)"
	if pgConn.ParameterStatus("crdb_version") != "" {
		copySql = "COPY foo FROM STDIN WITH CSV"
	}
	ct, err := pgConn.CopyFrom(ctx, r, copySql)
	cancel()
	assert.Equal(t, int64(0), ct.RowsAffected())
	assert.Error(t, err)

	assert.True(t, pgConn.IsClosed())
	select {
	case <-pgConn.CleanupDone():
	case <-time.After(5 * time.Second):
		t.Fatal("Connection cleanup exceeded maximum time")
	}
}

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