[feat] RR Create API 3: Add DAO for interacting with SMT metadata tables

.github/CODEOWNERS

@@ -0,0 +1,3 @@
+# review when someone opens a pull request
+
+* @GoogleCloudPlatform/spanner-migrations-team

accessors/clients/dataflow/dataflow_client.go

@@ -19,19 +19,13 @@ import (
 	"sync"
 
 	dataflow "cloud.google.com/go/dataflow/apiv1beta3"
-	"cloud.google.com/go/dataflow/apiv1beta3/dataflowpb"
-	"github.com/googleapis/gax-go/v2"
 )
 
-type DataflowClient interface {
-	LaunchFlexTemplate(ctx context.Context, req *dataflowpb.LaunchFlexTemplateRequest, opts ...gax.CallOption) (*dataflowpb.LaunchFlexTemplateResponse, error)
-}
-
 var once sync.Once
 var dfClient *dataflow.FlexTemplatesClient
 
 // This function is declared as a global variable to make it testable. The unit
-// tests edit this function, acting like a double.
+// tests update this function, acting like a double.
 var newFlexTemplatesClient = dataflow.NewFlexTemplatesClient
 
 func GetOrCreateClient(ctx context.Context) (*dataflow.FlexTemplatesClient, error) {

accessors/clients/dataflow/interface.go

@@ -0,0 +1,44 @@
+// Copyright 2024 Google LLC
+//
+// 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 dataflowclient
+
+import (
+	"context"
+
+	dataflow "cloud.google.com/go/dataflow/apiv1beta3"
+	"cloud.google.com/go/dataflow/apiv1beta3/dataflowpb"
+	"github.com/googleapis/gax-go/v2"
+)
+
+// Use this interface instead of dataflow.FlexTemplatesClient to support mocking.
+type DataflowClient interface {
+	LaunchFlexTemplate(ctx context.Context, req *dataflowpb.LaunchFlexTemplateRequest, opts ...gax.CallOption) (*dataflowpb.LaunchFlexTemplateResponse, error)
+}
+
+// This implements the DataflowClient interface. This is the primary implementation that should be used in all places other than tests.
+type DataflowClientImpl struct {
+	client *dataflow.FlexTemplatesClient
+}
+
+func NewDataflowClientImpl(ctx context.Context) (*DataflowClientImpl, error) {
+	c, err := GetOrCreateClient(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &DataflowClientImpl{client: c}, nil
+}
+
+func (c *DataflowClientImpl) LaunchFlexTemplate(ctx context.Context, req *dataflowpb.LaunchFlexTemplateRequest, opts ...gax.CallOption) (*dataflowpb.LaunchFlexTemplateResponse, error) {
+	return c.client.LaunchFlexTemplate(ctx, req, opts...)
+}

accessors/clients/dataflow/mocks.go

@@ -0,0 +1,31 @@
+// Copyright 2024 Google LLC
+//
+// 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 dataflowclient
+
+import (
+	"context"
+
+	"cloud.google.com/go/dataflow/apiv1beta3/dataflowpb"
+	"github.com/googleapis/gax-go/v2"
+)
+
+// Mock that implements the DataflowClient interface.
+// Pass in unit tests where DataflowClient is an input parameter.
+type DataflowClientMock struct {
+	LaunchFlexTemplateMock func(ctx context.Context, req *dataflowpb.LaunchFlexTemplateRequest, opts ...gax.CallOption) (*dataflowpb.LaunchFlexTemplateResponse, error)
+}
+
+func (dcm *DataflowClientMock) LaunchFlexTemplate(ctx context.Context, req *dataflowpb.LaunchFlexTemplateRequest, opts ...gax.CallOption) (*dataflowpb.LaunchFlexTemplateResponse, error) {
+	return dcm.LaunchFlexTemplateMock(ctx, req, opts...)
+}

accessors/clients/datastream/datastream_client.go

@@ -0,0 +1,43 @@
+// Copyright 2024 Google LLC
+//
+// 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 datastreamclient
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	datastream "cloud.google.com/go/datastream/apiv1"
+)
+
+var once sync.Once
+var dsClient *datastream.Client
+
+// This function is declared as a global variable to make it testable. The unit
+// tests update this function, acting like a double.
+var newClient = datastream.NewClient
+
+func GetOrCreateClient(ctx context.Context) (*datastream.Client, error) {
+	var err error
+	if dsClient == nil {
+		once.Do(func() {
+			dsClient, err = newClient(ctx)
+		})
+		if err != nil {
+			return nil, fmt.Errorf("failed to create datastream client: %v", err)
+		}
+		return dsClient, nil
+	}
+	return dsClient, nil
+}

accessors/clients/datastream/datastream_client_test.go

@@ -0,0 +1,119 @@
+// Copyright 2024 Google LLC
+//
+// 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 datastreamclient
+
+// TODO: Currently this test is intrusive and not using any accessors to mutate the code under test.
+//       Freeze on the right pattern and fork this into the test package.
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+
+	datastream "cloud.google.com/go/datastream/apiv1"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/logger"
+	"github.com/stretchr/testify/assert"
+	"go.uber.org/zap"
+	"google.golang.org/api/option"
+)
+
+func init() {
+	logger.Log = zap.NewNop()
+}
+
+func TestMain(m *testing.M) {
+	res := m.Run()
+	os.Exit(res)
+}
+
+func resetTest() {
+	dsClient = nil
+	once = sync.Once{}
+}
+
+func TestGetOrCreateClient_Basic(t *testing.T) {
+	resetTest()
+	ctx := context.Background()
+	oldFunc := newClient
+	defer func() { newClient = oldFunc }()
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return &datastream.Client{}, nil
+	}
+	c, err := GetOrCreateClient(ctx)
+	assert.NotNil(t, c)
+	assert.Nil(t, err)
+}
+
+func TestGetOrCreateClient_OnlyOnceViaSync(t *testing.T) {
+	resetTest()
+	ctx := context.Background()
+	oldFunc := newClient
+	defer func() { newClient = oldFunc }()
+
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return &datastream.Client{}, nil
+	}
+	c, err := GetOrCreateClient(ctx)
+	assert.NotNil(t, c)
+	assert.Nil(t, err)
+	// Explicitly set the client to nil. Running GetOrCreateClient should not create a
+	// new client since sync would already be executed.
+	dsClient = nil
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return nil, fmt.Errorf("test error")
+	}
+	c, err = GetOrCreateClient(ctx)
+	assert.Nil(t, c)
+	assert.Nil(t, err)
+}
+
+func TestGetOrCreateClient_OnlyOnceViaIf(t *testing.T) {
+	resetTest()
+	ctx := context.Background()
+	oldFunc := newClient
+	defer func() { newClient = oldFunc }()
+
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return &datastream.Client{}, nil
+	}
+	oldC, err := GetOrCreateClient(ctx)
+	assert.NotNil(t, oldC)
+	assert.Nil(t, err)
+
+	// Explicitly reset once. Running GetOrCreateClient should not create a
+	// new client the if condition should prevent it.
+	once = sync.Once{}
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return nil, fmt.Errorf("test error")
+	}
+	newC, err := GetOrCreateClient(ctx)
+	assert.Equal(t, oldC, newC)
+	assert.Nil(t, err)
+}
+
+func TestGetOrCreateClient_Error(t *testing.T) {
+	resetTest()
+	ctx := context.Background()
+	oldFunc := newClient
+	defer func() { newClient = oldFunc }()
+
+	newClient = func(ctx context.Context, opts ...option.ClientOption) (*datastream.Client, error) {
+		return nil, fmt.Errorf("test error")
+	}
+	c, err := GetOrCreateClient(ctx)
+	assert.Nil(t, c)
+	assert.NotNil(t, err)
+}

accessors/clients/datastream/datastream_test/mocks.go

@@ -0,0 +1,46 @@
+// Copyright 2024 Google LLC
+//
+// 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 datastreamclient_test
+
+import (
+	"context"
+
+	datastreampb "cloud.google.com/go/datastream/apiv1/datastreampb"
+	"github.com/GoogleCloudPlatform/spanner-migration-tool/accessors/clients/operation"
+	"github.com/googleapis/gax-go/v2"
+	"github.com/stretchr/testify/mock"
+)
+
+// Mock that implements the DatastreamClient interface.
+// Pass in unit tests where DatastreamClient is an input parameter.
+type DatastreamClientMock struct {
+	mock.Mock
+}
+
+func (m *DatastreamClientMock) CreateStream(ctx context.Context, req *datastreampb.CreateStreamRequest, opts ...gax.CallOption) (*operation.OperationWrapper[datastreampb.Stream], error) {
+	args := m.Called(ctx, req, opts)
+	// Avoid panic for typeassertion due to null pointer.
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*operation.OperationWrapper[datastreampb.Stream]), args.Error(1)
+}
+func (m *DatastreamClientMock) UpdateStream(ctx context.Context, req *datastreampb.UpdateStreamRequest, opts ...gax.CallOption) (*operation.OperationWrapper[datastreampb.Stream], error) {
+	args := m.Called(ctx, req, opts)
+	// Avoid panic for typeassertion due to null pointer.
+	if args.Get(0) == nil {
+		return nil, args.Error(1)
+	}
+	return args.Get(0).(*operation.OperationWrapper[datastreampb.Stream]), args.Error(1)
+}