Repository: cloudspannerecosystem/harbourbridge
Pull request #170 (2021-09-16 11:30:10)

Refactored context and database client
Changed parameters to pass context and database client rather than generating a new one each time.

Also modified parameters to use dbURI wherever possible rather than constructing a new one using project, instance and dbname.

agasheesh: I will wait for https://github.com/cloudspannerecosystem/harbourbridge/pull/163 to be re-submitted with fixes, before reviewing this PR since the diff is kind of incorrect at right now.

Patch:
diff --git a/README.md b/README.md
index 710d1eb..b5289a9 100644
--- a/README.md
+++ b/README.md
@@ -105,7 +105,7 @@ to set up your local development environment with authentication credentials.
 Set the GCLOUD_PROJECT environment variable to your Google Cloud project ID:
 
 ```sh
-export GCLOUD_PROJECT=[MY_PROJECT_ID]
+export GCLOUD_PROJECT=my-project-id
 ```
 
 If you do not already have a Cloud Spanner instance, or you want to use a
@@ -178,7 +178,7 @@ conversion), a session file and a bad data file (if any data was dropped).
 If you don't have ready access to a PostgreSQL or MySQL database, some example
 dump files can be found [here](examples). The files
 [cart.pg_dump](examples/cart.pg_dump) and
-[cart.mysqldump](examples/cart.mysqldump] contain pg_dump and mysqldump output
+[cart.mysqldump](examples/cart.mysqldump) contain pg_dump and mysqldump output
 for a very basic shopping cart application (just two tables, one for products
 and one for user carts). The files [singers.pg_dump](examples/singers.pg_dump)
 and [singers.mysqldump](examples/singers.mysqldump) contain pg_dump and
diff --git a/cmd/cmd.go b/cmd/cmd.go
index e94ce3c..252f262 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -16,9 +16,11 @@
 package cmd
 
 import (
+	"context"
 	"fmt"
 	"time"
 
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
@@ -36,10 +38,9 @@ var (
 // 2. Create database (if schemaOnly is set to false)
 // 3. Run data conversion (if schemaOnly is set to false)
 // 4. Generate report
-func CommandLine(driver, targetDb, projectID, instanceID, dbName string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *conversion.IOStreams, outputFilePrefix string, now time.Time) error {
+func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *conversion.IOStreams, outputFilePrefix string, now time.Time) error {
 	var conv *internal.Conv
 	var err error
-
 	if !dataOnly {
 		conv, err = conversion.SchemaConv(driver, targetDb, ioHelper, schemaSampleSize)
 		if err != nil {
@@ -62,13 +63,17 @@ func CommandLine(driver, targetDb, projectID, instanceID, dbName string, dataOnl
 			return err
 		}
 	}
-
-	dbURI, err := conversion.CreateOrUpdateDatabase(projectID, instanceID, dbName, conv, ioHelper.Out)
+	adminClient, err := database.NewDatabaseAdminClient(ctx)
+	if err != nil {
+		return fmt.Errorf("can't create admin client: %w", conversion.AnalyzeError(err, dbURI))
+	}
+	defer adminClient.Close()
+	err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, conv, ioHelper.Out)
 	if err != nil {
 		return fmt.Errorf("can't create/update database: %v", err)
 	}
 
-	client, err := conversion.GetClient(dbURI)
+	client, err := conversion.GetClient(ctx, dbURI)
 	if err != nil {
 		return fmt.Errorf("can't create client for db %s: %v", dbURI, err)
 	}
@@ -78,7 +83,7 @@ func CommandLine(driver, targetDb, projectID, instanceID, dbName string, dataOnl
 		return fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 	}
 	if !skipForeignKeys {
-		if err = conversion.UpdateDDLForeignKeys(projectID, instanceID, dbName, conv, ioHelper.Out); err != nil {
+		if err = conversion.UpdateDDLForeignKeys(ctx, adminClient, dbURI, conv, ioHelper.Out); err != nil {
 			return fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
 		}
 	}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index 3a64373..0efc092 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -436,14 +436,7 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 }
 
 // VerifyDb checks whether the db exists and if it does, verifies if the schema is what we currently support.
-func VerifyDb(project, instance, dbName string) (dbExists bool, err error) {
-	ctx := context.Background()
-	adminClient, err := database.NewDatabaseAdminClient(ctx)
-	if err != nil {
-		return false, fmt.Errorf("can't create admin client: %w", analyzeError(err, project, instance))
-	}
-	defer adminClient.Close()
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
+func VerifyDb(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string) (dbExists bool, err error) {
 	dbExists, err = CheckExistingDb(ctx, adminClient, dbURI)
 	if err != nil {
 		return dbExists, err
@@ -479,47 +472,32 @@ func ValidateDDL(ctx context.Context, adminClient *database.DatabaseAdminClient,
 	return nil
 }
 
-func CreateOrUpdateDatabase(project, instance, dbName string, conv *internal.Conv, out *os.File) (dbURI string, err error) {
-	ctx := context.Background()
-	adminClient, err := database.NewDatabaseAdminClient(ctx)
+func CreateOrUpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+	dbExists, err := VerifyDb(ctx, adminClient, dbURI)
 	if err != nil {
-		return "", fmt.Errorf("can't create admin client: %w", analyzeError(err, project, instance))
-	}
-	defer adminClient.Close()
-
-	dbExists, err := VerifyDb(project, instance, dbName)
-	if err != nil {
-		return "", err
+		return err
 	}
 	if dbExists {
-		dbURI, err = UpdateDatabase(project, instance, dbName, conv, out)
+		err := UpdateDatabase(ctx, adminClient, dbURI, conv, out)
 		if err != nil {
-			fmt.Printf("\nCan't update database schema: %v\n", err)
-			return "", fmt.Errorf("can't update database schema")
+			return fmt.Errorf("can't update database schema: %v", err)
 		}
-		return dbURI, nil
 	} else {
-		dbURI, err = CreateDatabase(project, instance, dbName, conv, out)
+		err := CreateDatabase(ctx, adminClient, dbURI, conv, out)
 		if err != nil {
-			fmt.Printf("\nCan't create database: %v\n", err)
-			return "", fmt.Errorf("can't create database")
+			return fmt.Errorf("can't create database: %v", err)
 		}
-		return dbURI, nil
 	}
+	return nil
 }
 
 // CreateDatabase returns a newly create Spanner DB.
 // It automatically determines an appropriate project, selects a
 // Spanner instance to use, generates a new Spanner DB name,
 // and call into the Spanner admin interface to create the new DB.
-func CreateDatabase(project, instance, dbName string, conv *internal.Conv, out *os.File) (string, error) {
+func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+	project, instance, dbName := parseDbURI(dbURI)
 	fmt.Fprintf(out, "Creating new database %s in instance %s with default permissions ... \n", dbName, instance)
-	ctx := context.Background()
-	adminClient, err := NewDatabaseAdminClient(ctx)
-	if err != nil {
-		return "", fmt.Errorf("can't create admin client: %w", analyzeError(err, project, instance))
-	}
-	defer adminClient.Close()
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
 	// using backticks (to avoid any issues with Spanner reserved words).
@@ -531,26 +509,17 @@ func CreateDatabase(project, instance, dbName string, conv *internal.Conv, out *
 		ExtraStatements: schema,
 	})
 	if err != nil {
-		return "", fmt.Errorf("can't build CreateDatabaseRequest: %w", analyzeError(err, project, instance))
+		return fmt.Errorf("can't build CreateDatabaseRequest: %w", AnalyzeError(err, dbURI))
 	}
 	if _, err := op.Wait(ctx); err != nil {
-		return "", fmt.Errorf("createDatabase call failed: %w", analyzeError(err, project, instance))
+		return fmt.Errorf("createDatabase call failed: %w", AnalyzeError(err, dbURI))
 	}
 	fmt.Fprintf(out, "Created database successfully.\n")
-	return fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName), nil
+	return nil
 }
 
-// UpdateDatabase updates the DDL of the spanner DB and return the path.
-func UpdateDatabase(project, instance, dbName string, conv *internal.Conv, out *os.File) (string, error) {
-	fmt.Fprintf(out, "Updating schema for database %s in instance %s ... \n", dbName, instance)
-	ctx := context.Background()
-	adminClient, err := database.NewDatabaseAdminClient(ctx)
-	if err != nil {
-		return "", fmt.Errorf("can't create admin client: %w", analyzeError(err, project, instance))
-	}
-	defer adminClient.Close()
-
-	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
+func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
+	fmt.Fprintf(out, "Updating schema for %s with default permissions ... \n", dbURI)
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
 	// using backticks (to avoid any issues with Spanner reserved words).
@@ -560,25 +529,51 @@ func UpdateDatabase(project, instance, dbName string, conv *internal.Conv, out *
 		Statements: conv.SpSchema.GetDDL(ddl.Config{Comments: false, ProtectIds: true, Tables: true, ForeignKeys: false}),
 	})
 	if err != nil {
-		return "", fmt.Errorf("can't submit update schema request: %s", err)
+		return fmt.Errorf("can't build UpdateDatabaseDdlRequest: %w", AnalyzeError(err, dbURI))
 	}
 	if err := op.Wait(ctx); err != nil {
-		return "", fmt.Errorf("can't update schema statement: %s", err)
+		return fmt.Errorf("UpdateDatabaseDdl call failed: %w", AnalyzeError(err, dbURI))
 	}
 	fmt.Fprintf(out, "Updated schema successfully.\n")
-	return dbURI, nil
+	return nil
+}
+
+// parseURI parses an unknown URI string that could be a database, instance or project URI.
+func parseURI(URI string) (project, instance, dbName string) {
+	project, instance, dbName = "", "", ""
+	if strings.Contains(URI, "databases") {
+		project, instance, dbName = parseDbURI(URI)
+	} else if strings.Contains(URI, "instances") {
+		project, instance = parseInstanceURI(URI)
+	} else if strings.Contains(URI, "projects") {
+		project = parseProjectURI(URI)
+	}
+	return
+}
+
+func parseDbURI(dbURI string) (project, instance, dbName string) {
+	split := strings.Split(dbURI, "/databases/")
+	project, instance = parseInstanceURI(split[0])
+	dbName = split[1]
+	return
+}
+
+func parseInstanceURI(instanceURI string) (project, instance string) {
+	split := strings.Split(instanceURI, "/instances/")
+	project = parseProjectURI(split[0])
+	instance = split[1]
+	return
+}
+
+func parseProjectURI(projectURI string) (project string) {
+	split := strings.Split(projectURI, "/")
+	project = split[1]
+	return
 }
 
 // UpdateDDLForeignKeys updates the Spanner database with foreign key
 // constraints using ALTER TABLE statements.
-func UpdateDDLForeignKeys(project, instance, dbName string, conv *internal.Conv, out *os.File) error {
-	ctx := context.Background()
-	adminClient, err := NewDatabaseAdminClient(ctx)
-	if err != nil {
-		return fmt.Errorf("can't create admin client: %w", analyzeError(err, project, instance))
-	}
-	defer adminClient.Close()
-
+func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
 	// using backticks (to avoid any issues with Spanner reserved words).
@@ -586,7 +581,7 @@ func UpdateDDLForeignKeys(project, instance, dbName string, conv *internal.Conv,
 	if len(fkStmts) == 0 {
 		return nil
 	}
-	msg := fmt.Sprintf("Updating schema of database %s in instance %s with foreign key constraints ...", dbName, instance)
+	msg := fmt.Sprintf("Updating schema of database %s with foreign key constraints ...", dbURI)
 	p := internal.NewProgress(int64(len(fkStmts)), msg, internal.Verbose())
 
 	workers := make(chan int, MaxWorkers)
@@ -611,9 +606,9 @@ func UpdateDDLForeignKeys(project, instance, dbName string, conv *internal.Conv,
 				progressMutex.Unlock()
 				workers <- workerId
 			}()
-			internal.VerbosePrintf("Submitting new FK create request for %s: %s\n", dbName, fkStmt)
+			internal.VerbosePrintf("Submitting new FK create request: %s\n", fkStmt)
 			op, err := adminClient.UpdateDatabaseDdl(ctx, &adminpb.UpdateDatabaseDdlRequest{
-				Database:   fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName),
+				Database:   dbURI,
 				Statements: []string{fkStmt},
 			})
 			if err != nil {
@@ -657,8 +652,8 @@ func GetProject() (string, error) {
 // GetInstance returns the Spanner instance we should use for creating DBs.
 // If the user specified instance (via flag 'instance') then use that.
 // Otherwise try to deduce the instance using gcloud.
-func GetInstance(project string, out *os.File) (string, error) {
-	l, err := getInstances(project)
+func GetInstance(ctx context.Context, project string, out *os.File) (string, error) {
+	l, err := getInstances(ctx, project)
 	if err != nil {
 		return "", err
 	}
@@ -683,11 +678,10 @@ func GetInstance(project string, out *os.File) (string, error) {
 		"Please use the flag '--instance' to select an instance", project)
 }
 
-func getInstances(project string) ([]string, error) {
-	ctx := context.Background()
-	instanceClient, err := NewInstanceAdminClient(ctx)
+func getInstances(ctx context.Context, project string) ([]string, error) {
+	instanceClient, err := instance.NewInstanceAdminClient(ctx)
 	if err != nil {
-		return nil, analyzeError(err, project, "")
+		return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
 	}
 	it := instanceClient.ListInstances(ctx, &instancepb.ListInstancesRequest{Parent: fmt.Sprintf("projects/%s", project)})
 	var l []string
@@ -697,7 +691,7 @@ func getInstances(project string) ([]string, error) {
 			break
 		}
 		if err != nil {
-			return nil, analyzeError(err, project, "")
+			return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
 		}
 		l = append(l, strings.TrimPrefix(resp.Name, fmt.Sprintf("projects/%s/instances/", project)))
 	}
@@ -879,9 +873,10 @@ func getPassword() string {
 	return strings.TrimSpace(string(bytePassword))
 }
 
-// analyzeError inspects an error returned from Cloud Spanner and adds information
+// AnalyzeError inspects an error returned from Cloud Spanner and adds information
 // about potential root causes e.g. authentication issues.
-func analyzeError(err error, project, instance string) error {
+func AnalyzeError(err error, URI string) error {
+	project, instance, _ := parseURI(URI)
 	e := strings.ToLower(err.Error())
 	if containsAny(e, []string{"unauthenticated", "cannot fetch token", "default credentials"}) {
 		return fmt.Errorf("%w.\n"+`
@@ -954,8 +949,7 @@ func NewSpannerClient(ctx context.Context, db string) (*sp.Client, error) {
 }
 
 // GetClient returns a new Spanner client.  It uses the background context.
-func GetClient(db string) (*sp.Client, error) {
-	ctx := context.Background()
+func GetClient(ctx context.Context, db string) (*sp.Client, error) {
 	return NewSpannerClient(ctx, db)
 }
 
diff --git a/main.go b/main.go
index c95f05e..2d9ee05 100644
--- a/main.go
+++ b/main.go
@@ -141,7 +141,7 @@ func main() {
 
 		instance = instanceOverride
 		if instance == "" {
-			instance, err = conversion.GetInstance(project, ioHelper.Out)
+			instance, err = conversion.GetInstance(ctx, project, ioHelper.Out)
 			if err != nil {
 				fmt.Printf("\nCan't get instance: %v\n", err)
 				panic(fmt.Errorf("can't get instance"))
@@ -160,6 +160,7 @@ func main() {
 			panic(fmt.Errorf("can't get database name"))
 		}
 	}
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
 
 	// If filePrefix not explicitly set, use dbName.
 	if filePrefix == "" {
@@ -168,7 +169,7 @@ func main() {
 
 	// TODO (agasheesh@): Collect all the config state in a single struct and pass the same to CommandLine instead of
 	// passing multiple parameters. Config state would be populated by parsing the flags and environment variables.
-	err = cmd.CommandLine(driverName, targetDb, project, instance, dbName, dataOnly, schemaOnly, skipForeignKeys, schemaSampleSize, sessionJSON, &ioHelper, filePrefix, now)
+	err = cmd.CommandLine(ctx, driverName, targetDb, dbURI, dataOnly, schemaOnly, skipForeignKeys, schemaSampleSize, sessionJSON, &ioHelper, filePrefix, now)
 	if err != nil {
 		panic(err)
 	}


Test patch:
diff --git a/testing/conversion/conversion_test.go b/testing/conversion/conversion_test.go
index 888bf85..f3c0c57 100644
--- a/testing/conversion/conversion_test.go
+++ b/testing/conversion/conversion_test.go
@@ -184,19 +184,20 @@ func TestUpdateDDLForeignKeys(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
+		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
 		conv := BuildConv(t, tc.numCols, tc.numFks, false)
-		dbpath, err := conversion.CreateDatabase(projectID, instanceID, tc.dbName, conv, os.Stdout)
+		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, conv, os.Stdout)
 		if err != nil {
 			t.Fatal(err)
 		}
 		conversion.MaxWorkers = tc.numWorkers
-		if err = conversion.UpdateDDLForeignKeys(projectID, instanceID, tc.dbName, conv, os.Stdout); err != nil {
+		if err = conversion.UpdateDDLForeignKeys(ctx, databaseAdmin, dbURI, conv, os.Stdout); err != nil {
 			t.Fatalf("\nCan't perform update operation on db %s with foreign keys: %v\n", tc.dbName, err)
 		}
 
-		checkResults(t, dbpath, tc.numFks)
+		checkResults(t, dbURI, tc.numFks)
 		// Drop the database later.
-		defer dropDatabase(t, dbpath)
+		defer dropDatabase(t, dbURI)
 	}
 }
 
@@ -214,13 +215,14 @@ func TestVerifyDb(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
+		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
 		if tc.dbExists {
-			dbURI, err := conversion.CreateDatabase(projectID, instanceID, tc.dbName, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
+			err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
 			if err != nil {
 				t.Fatal(err)
 			}
 			defer dropDatabase(t, dbURI)
-			dbExists, err := conversion.VerifyDb(projectID, instanceID, tc.dbName)
+			dbExists, err := conversion.VerifyDb(ctx, databaseAdmin, dbURI)
 			assert.True(t, dbExists)
 			if tc.emptySchema {
 				assert.Nil(t, err)
@@ -228,7 +230,7 @@ func TestVerifyDb(t *testing.T) {
 				assert.NotNil(t, err)
 			}
 		} else {
-			dbExists, err := conversion.VerifyDb(projectID, instanceID, tc.dbName)
+			dbExists, err := conversion.VerifyDb(ctx, databaseAdmin, dbURI)
 			assert.Nil(t, err)
 			assert.False(t, dbExists)
 		}
@@ -238,8 +240,8 @@ func TestVerifyDb(t *testing.T) {
 
 func TestCheckExistingDb(t *testing.T) {
 	onlyRunForEmulatorTest(t)
-
-	dbURI, err := conversion.CreateDatabase(projectID, instanceID, "check-db-exists", internal.MakeConv(), os.Stdout)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, "check-db-exists")
+	err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, internal.MakeConv(), os.Stdout)
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -271,7 +273,8 @@ func TestValidateDDL(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		dbURI, err := conversion.CreateDatabase(projectID, instanceID, tc.dbName, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
+		dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, tc.dbName)
+		err := conversion.CreateDatabase(ctx, databaseAdmin, dbURI, BuildConv(t, 2, 0, tc.emptySchema), os.Stdout)
 		if err != nil {
 			t.Fatal(err)
 		}
diff --git a/testing/dynamodb/integration_test.go b/testing/dynamodb/integration_test.go
index 6f51f9b..b8e5c2c 100644
--- a/testing/dynamodb/integration_test.go
+++ b/testing/dynamodb/integration_test.go
@@ -46,6 +46,7 @@ var (
 	projectID  string
 	instanceID string
 
+	ctx           context.Context
 	databaseAdmin *database.DatabaseAdminClient
 	dydbClient    *dydb.DynamoDB
 )
@@ -88,7 +89,7 @@ func initIntegrationTests() (cleanup func()) {
 	projectID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_PROJECT_ID")
 	instanceID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_INSTANCE_ID")
 
-	ctx := context.Background()
+	ctx = context.Background()
 	flag.Parse() // Needed for testing.Short().
 	noop := func() {}
 
@@ -124,12 +125,12 @@ func initIntegrationTests() (cleanup func()) {
 	}
 }
 
-func dropDatabase(t *testing.T, dbPath string) {
+func dropDatabase(t *testing.T, dbURI string) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	// Drop the testing database.
-	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbPath}); err != nil {
-		t.Fatalf("failed to drop testing database %v: %v", dbPath, err)
+	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbURI}); err != nil {
+		t.Fatalf("failed to drop testing database %v: %v", dbURI, err)
 	}
 
 	deleteTableInput := &dydb.DeleteTableInput{
@@ -213,23 +214,22 @@ func TestIntegration_DYNAMODB_SimpleUse(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.DYNAMODB, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	err := cmd.CommandLine(conversion.DYNAMODB, "spanner", projectID, instanceID, dbName, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
+	err := cmd.CommandLine(ctx, conversion.DYNAMODB, "spanner", dbURI, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
-func checkResults(t *testing.T, dbPath string) {
+func checkResults(t *testing.T, dbURI string) {
 	// Make a query to check results.
-	ctx := context.Background()
-	client, err := spanner.NewClient(ctx, dbPath)
+	client, err := spanner.NewClient(ctx, dbURI)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index a47ae3b..ff4ceec 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -41,6 +41,7 @@ var (
 	projectID  string
 	instanceID string
 
+	ctx           context.Context
 	databaseAdmin *database.DatabaseAdminClient
 )
 
@@ -55,7 +56,7 @@ func initIntegrationTests() (cleanup func()) {
 	projectID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_PROJECT_ID")
 	instanceID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_INSTANCE_ID")
 
-	ctx := context.Background()
+	ctx = context.Background()
 	flag.Parse() // Needed for testing.Short().
 	noop := func() {}
 
@@ -85,12 +86,12 @@ func initIntegrationTests() (cleanup func()) {
 	}
 }
 
-func dropDatabase(t *testing.T, dbPath string) {
+func dropDatabase(t *testing.T, dbURI string) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	// Drop the testing database.
-	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbPath}); err != nil {
-		t.Fatalf("failed to drop testing database %v: %v", dbPath, err)
+	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbURI}); err != nil {
+		t.Fatalf("failed to drop testing database %v: %v", dbURI, err)
 	}
 }
 
@@ -111,21 +112,21 @@ func TestIntegration_MYSQLDUMP_SimpleUse(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.MYSQLDUMP, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	dataFilepath := "../../test_data/mysqldump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 	f, err := os.Open(dataFilepath)
 	if err != nil {
 		t.Fatalf("failed to open the test data file: %v", err)
 	}
-	err = cmd.CommandLine(conversion.MYSQLDUMP, "spanner", projectID, instanceID, dbName, false, false, false, 0, "", &conversion.IOStreams{In: f, Out: os.Stdout}, filePrefix, now)
+	err = cmd.CommandLine(ctx, conversion.MYSQLDUMP, "spanner", dbURI, false, false, false, 0, "", &conversion.IOStreams{In: f, Out: os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func TestIntegration_MYSQL_SimpleUse(t *testing.T) {
@@ -137,17 +138,17 @@ func TestIntegration_MYSQL_SimpleUse(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.MYSQL, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	err := cmd.CommandLine(conversion.MYSQL, "spanner", projectID, instanceID, dbName, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
+	err := cmd.CommandLine(ctx, conversion.MYSQL, "spanner", dbURI, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func runSchemaOnly(t *testing.T, dbName, filePrefix, sessionFile, dumpFilePath string) {
@@ -217,10 +218,9 @@ func TestIntegration_MySQLDUMP_DataOnly(t *testing.T) {
 	checkResults(t, dbURI)
 }
 
-func checkResults(t *testing.T, dbPath string) {
+func checkResults(t *testing.T, dbURI string) {
 	// Make a query to check results.
-	ctx := context.Background()
-	client, err := spanner.NewClient(ctx, dbPath)
+	client, err := spanner.NewClient(ctx, dbURI)
 	if err != nil {
 		log.Fatal(err)
 	}
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index f718de3..55b67d5 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -41,6 +41,7 @@ var (
 	projectID  string
 	instanceID string
 
+	ctx           context.Context
 	databaseAdmin *database.DatabaseAdminClient
 )
 
@@ -55,7 +56,7 @@ func initIntegrationTests() (cleanup func()) {
 	projectID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_PROJECT_ID")
 	instanceID = os.Getenv("HARBOURBRIDGE_TESTS_GCLOUD_INSTANCE_ID")
 
-	ctx := context.Background()
+	ctx = context.Background()
 	flag.Parse() // Needed for testing.Short().
 	noop := func() {}
 
@@ -85,12 +86,12 @@ func initIntegrationTests() (cleanup func()) {
 	}
 }
 
-func dropDatabase(t *testing.T, dbPath string) {
+func dropDatabase(t *testing.T, dbURI string) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
 	defer cancel()
 	// Drop the testing database.
-	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbPath}); err != nil {
-		t.Fatalf("failed to drop testing database %v: %v", dbPath, err)
+	if err := databaseAdmin.DropDatabase(ctx, &databasepb.DropDatabaseRequest{Database: dbURI}); err != nil {
+		t.Fatalf("failed to drop testing database %v: %v", dbURI, err)
 	}
 }
 
@@ -113,21 +114,21 @@ func TestIntegration_PGDUMP_SimpleUse(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.PGDUMP, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	dataFilepath := "../../test_data/pg_dump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 	f, err := os.Open(dataFilepath)
 	if err != nil {
 		t.Fatalf("failed to open the test data file: %v", err)
 	}
-	err = cmd.CommandLine(conversion.PGDUMP, "spanner", projectID, instanceID, dbName, false, false, false, 0, "", &conversion.IOStreams{In: f, Out: os.Stdout}, filePrefix, now)
+	err = cmd.CommandLine(ctx, conversion.PGDUMP, "spanner", dbURI, false, false, false, 0, "", &conversion.IOStreams{In: f, Out: os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func TestIntegration_PGDUMP_Command(t *testing.T) {
@@ -138,7 +139,7 @@ func TestIntegration_PGDUMP_Command(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.PGDUMP, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 
 	dataFilepath := "../../test_data/pg_dump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
@@ -160,9 +161,9 @@ func TestIntegration_PGDUMP_Command(t *testing.T) {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func TestIntegration_PGDUMP_SchemaCommand(t *testing.T) {
@@ -197,17 +198,17 @@ func TestIntegration_POSTGRES_SimpleUse(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.POSTGRES, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
-	err := cmd.CommandLine(conversion.POSTGRES, "spanner", projectID, instanceID, dbName, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
+	err := cmd.CommandLine(ctx, conversion.POSTGRES, "spanner", dbURI, false, false, false, 0, "", &conversion.IOStreams{Out: os.Stdout}, filePrefix, now)
 	if err != nil {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func TestIntegration_POSTGRES_Command(t *testing.T) {
@@ -219,7 +220,7 @@ func TestIntegration_POSTGRES_Command(t *testing.T) {
 
 	now := time.Now()
 	dbName, _ := conversion.GetDatabaseName(conversion.POSTGRES, now)
-	dbPath := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
+	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
 	cmd := exec.Command("bash", "-c", fmt.Sprintf("go run github.com/cloudspannerecosystem/harbourbridge -instance %s -dbname %s -prefix %s -driver %s", instanceID, dbName, filePrefix, conversion.POSTGRES))
@@ -235,9 +236,9 @@ func TestIntegration_POSTGRES_Command(t *testing.T) {
 		t.Fatal(err)
 	}
 	// Drop the database later.
-	defer dropDatabase(t, dbPath)
+	defer dropDatabase(t, dbURI)
 
-	checkResults(t, dbPath)
+	checkResults(t, dbURI)
 }
 
 func TestIntegration_POSTGRES_SchemaCommand(t *testing.T) {
@@ -258,10 +259,9 @@ func TestIntegration_POSTGRES_SchemaCommand(t *testing.T) {
 	}
 }
 
-func checkResults(t *testing.T, dbPath string) {
+func checkResults(t *testing.T, dbURI string) {
 	// Make a query to check results.
-	ctx := context.Background()
-	client, err := spanner.NewClient(ctx, dbPath)
+	client, err := spanner.NewClient(ctx, dbURI)
 	if err != nil {
 		log.Fatal(err)
 	}