Merge branch 'main' into dapr-state-store-clickhouse

Author: middt

bindings/postgres/metadata.go

@@ -14,6 +14,7 @@ limitations under the License.
 package postgres
 
 import (
+	"errors"
 	"time"
 
 	"github.com/dapr/components-contrib/common/authentication/aws"
@@ -53,5 +54,9 @@ func (m *psqlMetadata) InitWithMetadata(meta map[string]string) error {
 		return err
 	}
 
+	if m.Timeout < 1*time.Second {
+		return errors.New("invalid value for 'timeout': must be greater than 1s")
+	}
+
 	return nil
 }

bindings/postgres/metadata_test.go

@@ -0,0 +1,88 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package postgres
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMetadata(t *testing.T) {
+	t.Run("missing connection string", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{}
+
+		err := m.InitWithMetadata(props)
+		require.Error(t, err)
+		require.ErrorContains(t, err, "connection string")
+	})
+
+	t.Run("has connection string", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{
+			"connectionString": "foo",
+		}
+
+		err := m.InitWithMetadata(props)
+		require.NoError(t, err)
+	})
+
+	t.Run("default timeout", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{
+			"connectionString": "foo",
+		}
+
+		err := m.InitWithMetadata(props)
+		require.NoError(t, err)
+		assert.Equal(t, 20*time.Second, m.Timeout)
+	})
+
+	t.Run("invalid timeout", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{
+			"connectionString": "foo",
+			"timeout":          "NaN",
+		}
+
+		err := m.InitWithMetadata(props)
+		require.Error(t, err)
+	})
+
+	t.Run("positive timeout", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{
+			"connectionString": "foo",
+			"timeout":          "42",
+		}
+
+		err := m.InitWithMetadata(props)
+		require.NoError(t, err)
+		assert.Equal(t, 42*time.Second, m.Timeout)
+	})
+
+	t.Run("zero timeout", func(t *testing.T) {
+		m := psqlMetadata{}
+		props := map[string]string{
+			"connectionString": "foo",
+			"timeout":          "0",
+		}
+
+		err := m.InitWithMetadata(props)
+		require.Error(t, err)
+	})
+}

bindings/postgres/postgres.go

@@ -73,11 +73,20 @@ func (p *Postgres) Init(ctx context.Context, meta bindings.Metadata) error {
 
 	// This context doesn't control the lifetime of the connection pool, and is
 	// only scoped to postgres creating resources at init.
-	p.db, err = pgxpool.NewWithConfig(ctx, poolConfig)
+	connCtx, connCancel := context.WithTimeout(ctx, m.Timeout)
+	defer connCancel()
+	p.db, err = pgxpool.NewWithConfig(connCtx, poolConfig)
 	if err != nil {
 		return fmt.Errorf("unable to connect to the DB: %w", err)
 	}
 
+	pingCtx, pingCancel := context.WithTimeout(ctx, m.Timeout)
+	defer pingCancel()
+	err = p.db.Ping(pingCtx)
+	if err != nil {
+		return fmt.Errorf("failed to ping the DB: %w", err)
+	}
+
 	return nil
 }
 

bindings/postgres/postgres_test.go

@@ -15,6 +15,7 @@ package postgres
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"testing"
@@ -62,6 +63,10 @@ func TestPostgresIntegration(t *testing.T) {
 		t.SkipNow()
 	}
 
+	t.Run("Test init configurations", func(t *testing.T) {
+		testInitConfiguration(t, url)
+	})
+
 	// live DB test
 	b := NewPostgres(logger.NewLogger("test")).(*Postgres)
 	m := bindings.Metadata{Base: metadata.Base{Properties: map[string]string{"connectionString": url}}}
@@ -131,6 +136,46 @@ func TestPostgresIntegration(t *testing.T) {
 	})
 }
 
+// testInitConfiguration tests valid and invalid config settings.
+func testInitConfiguration(t *testing.T, connectionString string) {
+	logger := logger.NewLogger("test")
+	tests := []struct {
+		name        string
+		props       map[string]string
+		expectedErr error
+	}{
+		{
+			name:        "Empty",
+			props:       map[string]string{},
+			expectedErr: errors.New("missing connection string"),
+		},
+		{
+			name:        "Valid connection string",
+			props:       map[string]string{"connectionString": connectionString},
+			expectedErr: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			p := NewPostgres(logger).(*Postgres)
+			defer p.Close()
+
+			metadata := bindings.Metadata{
+				Base: metadata.Base{Properties: tt.props},
+			}
+
+			err := p.Init(context.Background(), metadata)
+			if tt.expectedErr == nil {
+				require.NoError(t, err)
+			} else {
+				require.Error(t, err)
+				assert.Equal(t, tt.expectedErr, err)
+			}
+		})
+	}
+}
+
 func assertResponse(t *testing.T, res *bindings.InvokeResponse, err error) {
 	require.NoError(t, err)
 	assert.NotNil(t, res)

state/postgresql/v2/postgresql.go

@@ -99,16 +99,16 @@ func (p *PostgreSQL) Init(ctx context.Context, meta state.Metadata) (err error)
 	}
 
 	connCtx, connCancel := context.WithTimeout(ctx, p.metadata.Timeout)
+	defer connCancel()
 	p.db, err = pgxpool.NewWithConfig(connCtx, config)
-	connCancel()
 	if err != nil {
 		err = fmt.Errorf("failed to connect to the database: %w", err)
 		return err
 	}
 
 	pingCtx, pingCancel := context.WithTimeout(ctx, p.metadata.Timeout)
+	defer pingCancel()
 	err = p.db.Ping(pingCtx)
-	pingCancel()
 	if err != nil {
 		err = fmt.Errorf("failed to ping the database: %w", err)
 		return err