Move common ydb topics testing funcs to helpers

Move common ydb topics testing funcs to helpers. This will make it easier to work on other PRs about yds, ydb and lb
commit_hash:a43d0703b0d99bb6b8fa1b00c23adb64b334cb7f

.mapping.json

@@ -2909,6 +2909,7 @@
   "tests/helpers/utils.go":"transfer_manager/go/tests/helpers/utils.go",
   "tests/helpers/utils/test_read_closer.go":"transfer_manager/go/tests/helpers/utils/test_read_closer.go",
   "tests/helpers/ydb.go":"transfer_manager/go/tests/helpers/ydb.go",
+  "tests/helpers/ydb_topics/recipe.go":"transfer_manager/go/tests/helpers/ydb_topics/recipe.go",
   "tests/helpers/yt/yt_helpers.go":"transfer_manager/go/tests/helpers/yt/yt_helpers.go",
   "tests/large/docker-compose/README.md":"transfer_manager/go/tests/large/docker-compose/README.md",
   "tests/large/docker-compose/canondata/docker-compose.docker-compose.TestElasticToPgSnapshot/extracted":"transfer_manager/go/tests/large/docker-compose/canondata/docker-compose.docker-compose.TestElasticToPgSnapshot/extracted",

pkg/providers/ydb/source_test.go

@@ -15,9 +15,9 @@ import (
 	"github.com/doublecloud/transfer/library/go/test/canon"
 	"github.com/doublecloud/transfer/pkg/abstract"
 	"github.com/doublecloud/transfer/pkg/abstract/model"
+	recipe "github.com/doublecloud/transfer/tests/helpers/ydb_topics"
 	"github.com/stretchr/testify/require"
 	"github.com/ydb-platform/ydb-go-sdk/v3"
-	"github.com/ydb-platform/ydb-go-sdk/v3/sugar"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table/options"
 	"github.com/ydb-platform/ydb-go-sdk/v3/table/types"
@@ -41,7 +41,7 @@ func (s asyncSinkMock) Close() error {
 }
 
 func TestSourceCDC(t *testing.T) {
-	db := driver(t)
+	db := recipe.Driver(t)
 	transferID := "test_transfer"
 
 	srcCfgTemplate := YdbSource{
@@ -313,7 +313,7 @@ func waitExpectedEvents(t *testing.T, src *Source, expectedItemsCount int) []abs
 }
 
 func prepareTableAndFeed(t *testing.T, feedName, tableName string, differentKeysCount, updatesPerKey int) int {
-	db := driver(t)
+	db := recipe.Driver(t)
 	tablePath := formTablePath(tableName)
 	createTableAndFeed(t, db, feedName, tablePath,
 		options.WithColumn("id", types.Optional(types.TypeUint64)),
@@ -368,18 +368,3 @@ func createTableAndFeedWithMode(t *testing.T, db *ydb.Driver, feedName, tablePat
 func formTablePath(tableName string) string {
 	return "/" + path.Join(os.Getenv("YDB_DATABASE"), tableName)
 }
-
-func driver(t *testing.T) *ydb.Driver {
-	instance := os.Getenv("YDB_ENDPOINT")
-	database := os.Getenv("YDB_DATABASE")
-	token := os.Getenv("YDB_TOKEN")
-
-	db, err := ydb.Open(
-		context.Background(),
-		sugar.DSN(instance, database),
-		ydb.WithAccessTokenCredentials(token),
-	)
-	require.NoError(t, err)
-
-	return db
-}

tests/helpers/ydb_topics/recipe.go

@@ -0,0 +1,49 @@
+package recipe
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"github.com/ydb-platform/ydb-go-sdk/v3"
+	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
+	"github.com/ydb-platform/ydb-go-sdk/v3/sugar"
+)
+
+func Driver(t *testing.T, opts ...ydb.Option) *ydb.Driver {
+	instance, port, database, creds := InstancePortDatabaseCreds(t)
+	dsn := sugar.DSN(fmt.Sprintf("%s:%d", instance, port), database)
+	if creds != nil {
+		opts = append(opts, ydb.WithCredentials(creds))
+	}
+	driver, err := ydb.Open(context.Background(), dsn, opts...)
+	require.NoError(t, err)
+
+	return driver
+}
+
+func InstancePortDatabaseCreds(t *testing.T) (string, int, string, credentials.Credentials) {
+	parts := strings.Split(os.Getenv("YDB_ENDPOINT"), ":")
+	require.Len(t, parts, 2)
+
+	instance := parts[0]
+	port, err := strconv.Atoi(parts[1])
+	require.NoError(t, err)
+
+	database := os.Getenv("YDB_DATABASE")
+	if database == "" {
+		database = "local"
+	}
+
+	var creds credentials.Credentials
+	token := os.Getenv("YDB_TOKEN")
+	if token != "" {
+		creds = credentials.NewAccessTokenCredentials(token)
+	}
+
+	return instance, port, database, creds
+}