PG basic integration test

app/server/validate.go

@@ -107,7 +107,7 @@ func validateDataSourceInstance(logger *zap.Logger, dsi *api_common.TDataSourceI
 
 	switch dsi.GetKind() {
 	case api_common.EDataSourceKind_POSTGRESQL:
-		if dsi.GetPgOptions().Schema == "" {
+		if dsi.GetPgOptions().GetSchema() == "" {
 			return fmt.Errorf("schema field is empty: %w", common.ErrInvalidRequest)
 		}
 

tests/infra/datasource/clickhouse/datasource.go

@@ -53,10 +53,5 @@ func DeriveDataSourceFromDockerCompose(ed *docker_compose.EndpointDeterminer) (*
 		return nil, fmt.Errorf("derive HTTP endpoint: %w", err)
 	}
 
-	return datasource.NewDataSource(
-		map[api_common.EProtocol]*api_common.TDataSourceInstance{
-			api_common.EProtocol_HTTP:   dsiHTTP,
-			api_common.EProtocol_NATIVE: dsiNative,
-		},
-	), nil
+	return &datasource.DataSource{Instances: []*api_common.TDataSourceInstance{dsiHTTP, dsiNative}}, nil
 }

tests/infra/datasource/datasource.go

@@ -1,26 +1,9 @@
 package datasource
 
 import (
-	"fmt"
-
 	api_common "github.com/ydb-platform/fq-connector-go/api/common"
 )
 
 type DataSource struct {
-	dsi map[api_common.EProtocol]*api_common.TDataSourceInstance
-}
-
-func (ds *DataSource) GetDataSourceInstance(protocol api_common.EProtocol) (*api_common.TDataSourceInstance, error) {
-	result, exists := ds.dsi[protocol]
-	if !exists {
-		return nil, fmt.Errorf("unexpected protocol %v", protocol)
-	}
-
-	return result, nil
-}
-
-func NewDataSource(
-	dsi map[api_common.EProtocol]*api_common.TDataSourceInstance,
-) *DataSource {
-	return &DataSource{dsi: dsi}
+	Instances []*api_common.TDataSourceInstance
 }

tests/infra/datasource/docker-compose.yaml

@@ -20,3 +20,16 @@ services:
       nofile:
         soft: 262144
         hard: 262144
+
+  postgresql:
+    image: postgres
+    container_name: fq-connector-go-tests-postgresql
+    ports:
+      - 5432
+    environment:
+      POSTGRES_DB: connector
+      POSTGRES_USER: admin
+      POSTGRES_PASSWORD: password
+      PGDATA: /var/lib/postgresql/data/pgdata
+    volumes:
+      - ./postgresql/init:/docker-entrypoint-initdb.d

tests/infra/datasource/postgresql/datasource.go

@@ -0,0 +1,51 @@
+package postgresql
+
+import (
+	"fmt"
+
+	api_common "github.com/ydb-platform/fq-connector-go/api/common"
+	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource"
+	"github.com/ydb-platform/fq-connector-go/tests/infra/docker_compose"
+)
+
+const (
+	serviceName  = "postgresql"
+	internalPort = 5432
+	database     = "connector"
+	username     = "admin"
+	password     = "password"
+	schema       = "public"
+)
+
+func DeriveDataSourceFromDockerCompose(ed *docker_compose.EndpointDeterminer) (*datasource.DataSource, error) {
+	dsi := &api_common.TDataSourceInstance{
+		Kind:     api_common.EDataSourceKind_POSTGRESQL,
+		Database: database,
+		Credentials: &api_common.TCredentials{
+			Payload: &api_common.TCredentials_Basic{
+				Basic: &api_common.TCredentials_TBasic{
+					Username: username,
+					Password: password,
+				},
+			},
+		},
+		Protocol: api_common.EProtocol_NATIVE,
+		UseTls:   false,
+		Options: &api_common.TDataSourceInstance_PgOptions{
+			PgOptions: &api_common.TPostgreSQLDataSourceOptions{
+				Schema: schema,
+			},
+		},
+	}
+
+	var err error
+	dsi.Endpoint, err = ed.GetEndpoint(serviceName, internalPort)
+
+	if err != nil {
+		return nil, fmt.Errorf("derive endpoint: %w", err)
+	}
+
+	return &datasource.DataSource{
+		Instances: []*api_common.TDataSourceInstance{dsi},
+	}, nil
+}

tests/infra/datasource/postgresql/init/init_db.sh

@@ -0,0 +1,56 @@
+#!/bin/bash
+set -e
+
+psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
+    DROP TABLE IF EXISTS simple;
+    CREATE TABLE simple (id integer, col1 text, col2 integer);
+    INSERT INTO simple VALUES (1, 'pg_a', 10);
+    INSERT INTO simple VALUES (2, 'pg_b', 20);
+    INSERT INTO simple VALUES (3, 'pg_c', 30);
+    INSERT INTO simple VALUES (4, 'pg_d', 40);
+    INSERT INTO simple VALUES (5, 'pg_e', 50);
+EOSQL
+
+psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
+    DROP TABLE IF EXISTS primitives;
+    CREATE TABLE primitives (
+        col_01_bool bool,
+        col_02_smallint smallint,
+        col_03_int2 int2,
+        col_04_smallserial smallserial,
+        col_05_serial2 serial2,
+        col_06_integer integer,
+        col_07_int int,
+        col_08_int4 int4,
+        col_09_serial serial,
+        col_10_serial4 serial4,
+        col_11_bigint bigint,
+        col_12_int8 int8,
+        col_13_bigserial bigserial,
+        col_14_serial8 serial8,
+        col_15_real real,
+        col_16_float4 float4,
+        col_17_double_precision double precision,
+        col_18_float8 float8,
+        col_19_bytea bytea,
+        col_20_character_n character(20),
+        col_21_character_varying_n character varying(21),
+        col_22_text text,
+        col_23_timestamp timestamp,
+        col_24_date date
+    );
+    INSERT INTO primitives VALUES (
+        false, 2, 3, DEFAULT, DEFAULT, 6, 7, 8, DEFAULT, DEFAULT, 11, 12, DEFAULT, DEFAULT,
+        15.15, 16.16, 17.17, 18.18, 'az', 'az', 'az', 'az',
+        current_timestamp, current_timestamp);
+    INSERT INTO primitives VALUES (
+        true, -2, -3, DEFAULT, DEFAULT, -6, -7, -8, DEFAULT, DEFAULT, -11, -12, DEFAULT, DEFAULT,
+        -15.15, -16.16, -17.17, -18.18, 'буки', 'буки', 'буки', 'буки',
+        current_timestamp, current_timestamp);
+    INSERT INTO primitives VALUES (
+        NULL, NULL, NULL, DEFAULT, DEFAULT, NULL,
+        NULL, NULL, DEFAULT, DEFAULT, NULL, NULL,
+        DEFAULT, DEFAULT, NULL, NULL, NULL, NULL,
+        NULL, NULL, NULL, NULL, NULL, NULL
+        );
+EOSQL

tests/infra/datasource/postgresql/tables.go

@@ -0,0 +1,46 @@
+package postgresql
+
+import (
+	"github.com/ydb-platform/ydb-go-genproto/protos/Ydb"
+
+	api_service_protos "github.com/ydb-platform/fq-connector-go/api/service/protos"
+	"github.com/ydb-platform/fq-connector-go/app/common"
+	"github.com/ydb-platform/fq-connector-go/library/go/ptr"
+	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource"
+)
+
+var Tables = map[string]*datasource.Table{
+	"simple": {
+		SchemaYdb: &api_service_protos.TSchema{
+			Columns: []*Ydb.Column{
+				{
+					Name: "id",
+					Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_INT32)),
+				},
+				{
+					Name: "col1",
+					Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_UTF8)),
+				},
+				{
+					Name: "col2",
+					Type: common.MakeOptionalType(common.MakePrimitiveType(Ydb.Type_INT32)),
+				},
+			},
+		},
+		Records: []*datasource.Record{
+			{
+				Columns: []any{
+					[]*int32{ptr.Int32(1), ptr.Int32(2), ptr.Int32(3), ptr.Int32(4), ptr.Int32(5)},
+					[]*string{
+						ptr.String("pg_a"),
+						ptr.String("pg_b"),
+						ptr.String("pg_c"),
+						ptr.String("pg_d"),
+						ptr.String("pg_e"),
+					},
+					[]*int32{ptr.Int32(10), ptr.Int32(20), ptr.Int32(30), ptr.Int32(40), ptr.Int32(50)},
+				},
+			},
+		},
+	},
+}

tests/main_test.go

@@ -9,6 +9,7 @@ import (
 	testify_suite "github.com/stretchr/testify/suite"
 
 	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource/clickhouse"
+	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource/postgresql"
 	"github.com/ydb-platform/fq-connector-go/tests/suite"
 )
 
@@ -34,5 +35,24 @@ func TestSelectClickHouse(t *testing.T) {
 	ds, err := clickhouse.DeriveDataSourceFromDockerCompose(state.EndpointDeterminer)
 	require.NoError(t, err)
 
-	testify_suite.Run(t, NewSelectSuite(suite.NewBase(t, state, "SelectClickHouse"), ds))
+	testify_suite.Run(
+		t,
+		NewSelectSuite(
+			suite.NewBase(t, state, "SelectClickHouse"),
+			ds,
+			clickhouse.Tables),
+	)
+}
+
+func TestSelectPostgreSQL(t *testing.T) {
+	ds, err := postgresql.DeriveDataSourceFromDockerCompose(state.EndpointDeterminer)
+	require.NoError(t, err)
+
+	testify_suite.Run(
+		t,
+		NewSelectSuite(
+			suite.NewBase(t, state, "SelectPostgreSQL"),
+			ds,
+			postgresql.Tables),
+	)
 }

tests/select.go

@@ -12,7 +12,6 @@ import (
 	api_service_protos "github.com/ydb-platform/fq-connector-go/api/service/protos"
 	"github.com/ydb-platform/fq-connector-go/app/common"
 	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource"
-	"github.com/ydb-platform/fq-connector-go/tests/infra/datasource/clickhouse"
 	"github.com/ydb-platform/fq-connector-go/tests/suite"
 )
 
@@ -21,22 +20,19 @@ import (
 // * working with primitive and optional types
 type SelectSuite struct {
 	*suite.Base
-	protocols  []api_common.EProtocol
 	dataSource *datasource.DataSource
+	tables     map[string]*datasource.Table
 }
 
-func (s *SelectSuite) TestSimpleTable() {
-	for tableName, table := range clickhouse.Tables {
-		for _, protocol := range s.protocols {
-			s.doTestSimpleTable(tableName, table, protocol)
+func (s *SelectSuite) TestSimpleSelect() {
+	for _, dsi := range s.dataSource.Instances {
+		for tableName, table := range s.tables {
+			s.doTestSimpleSelect(tableName, table, dsi)
 		}
 	}
 }
 
-func (s *SelectSuite) doTestSimpleTable(tableName string, table *datasource.Table, protocol api_common.EProtocol) {
-	dsi, err := s.dataSource.GetDataSourceInstance(protocol)
-	s.Require().NoError(err)
-
+func (s *SelectSuite) doTestSimpleSelect(tableName string, table *datasource.Table, dsi *api_common.TDataSourceInstance) {
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 
@@ -48,7 +44,7 @@ func (s *SelectSuite) doTestSimpleTable(tableName string, table *datasource.Tabl
 	// describe table
 	describeTableResponse, err := s.Connector.Client().DescribeTable(ctx, dsi, typeMappingSettings, tableName)
 	s.Require().NoError(err)
-	s.Require().Equal(Ydb.StatusIds_SUCCESS, describeTableResponse.Error.Status)
+	s.Require().Equal(Ydb.StatusIds_SUCCESS, describeTableResponse.Error.Status, describeTableResponse.Error.String())
 	s.Require().True(
 		proto.Equal(table.SchemaYdb, describeTableResponse.Schema),
 		fmt.Sprintf("expected: %v\nactual:   %v\n", table.SchemaYdb, describeTableResponse.Schema),
@@ -79,14 +75,15 @@ func (s *SelectSuite) doTestSimpleTable(tableName string, table *datasource.Tabl
 	table.MatchRecords(s.T(), records)
 }
 
-func NewSelectSuite(baseSuite *suite.Base, dataSource *datasource.DataSource) *SelectSuite {
+func NewSelectSuite(
+	baseSuite *suite.Base,
+	dataSource *datasource.DataSource,
+	tables map[string]*datasource.Table,
+) *SelectSuite {
 	result := &SelectSuite{
-		Base: baseSuite,
-		protocols: []api_common.EProtocol{
-			api_common.EProtocol_HTTP,
-			api_common.EProtocol_NATIVE,
-		},
+		Base:       baseSuite,
 		dataSource: dataSource,
+		tables:     tables,
 	}
 
 	return result