DMS: Fix acceptance test failures

.changelog/33158.txt

@@ -0,0 +1,11 @@
+```release-note:enhancement
+resource/aws_dms_replication_instance: Add `network_type` argument
+```
+
+```release-note:enhancement
+data-source/aws_dms_replication_instance: Add `network_type` attribute
+```
+
+```release-note:enhancement
+data-source/aws_dms_endpoint: Fix crash when specified endpoint not found
+```

internal/service/dms/consts.go

@@ -6,6 +6,12 @@ package dms
 const (
 	endpointStatusDeleting = "deleting"
 
+	replicationInstanceStatusAvailable = "available"
+	replicationInstanceStatusCreating  = "creating"
+	replicationInstanceStatusDeleting  = "deleting"
+	replicationInstanceStatusModifying = "modifying"
+	replicationInstanceStatusUpgrading = "upgrading"
+
 	replicationTaskStatusCreating  = "creating"
 	replicationTaskStatusDeleting  = "deleting"
 	replicationTaskStatusFailed    = "failed"
@@ -125,3 +131,15 @@ func encryptionMode_Values() []string {
 		encryptionModeSseS3,
 	}
 }
+
+const (
+	networkTypeDual = "DUAL"
+	networkTypeIPv4 = "IPV4"
+)
+
+func networkType_Values() []string {
+	return []string{
+		networkTypeDual,
+		networkTypeIPv4,
+	}
+}

internal/service/dms/endpoint.go

@@ -17,6 +17,7 @@ import (
 	"github.com/hashicorp/go-cty/cty"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/customdiff"
+	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/retry"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/validation"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
@@ -959,10 +960,8 @@ func resourceEndpointRead(ctx context.Context, d *schema.ResourceData, meta inte
 		return sdkdiag.AppendErrorf(diags, "reading DMS Endpoint (%s): %s", d.Id(), err)
 	}
 
-	err = resourceEndpointSetState(d, endpoint)
-
-	if err != nil {
-		return sdkdiag.AppendErrorf(diags, "reading DMS Endpoint (%s): %s", d.Id(), err)
+	if err := resourceEndpointSetState(d, endpoint); err != nil {
+		return sdkdiag.AppendFromErr(diags, err)
 	}
 
 	return diags
@@ -2302,3 +2301,86 @@ func flattenTopLevelConnectionInfo(d *schema.ResourceData, endpoint *dms.Endpoin
 	d.Set("port", endpoint.Port)
 	d.Set("database_name", endpoint.DatabaseName)
 }
+
+func FindEndpointByID(ctx context.Context, conn *dms.DatabaseMigrationService, id string) (*dms.Endpoint, error) {
+	input := &dms.DescribeEndpointsInput{
+		Filters: []*dms.Filter{
+			{
+				Name:   aws.String("endpoint-id"),
+				Values: aws.StringSlice([]string{id}),
+			},
+		},
+	}
+
+	return findEndpoint(ctx, conn, input)
+}
+
+func findEndpoint(ctx context.Context, conn *dms.DatabaseMigrationService, input *dms.DescribeEndpointsInput) (*dms.Endpoint, error) {
+	output, err := findEndpoints(ctx, conn, input)
+
+	if err != nil {
+		return nil, err
+	}
+
+	return tfresource.AssertSinglePtrResult(output)
+}
+
+func findEndpoints(ctx context.Context, conn *dms.DatabaseMigrationService, input *dms.DescribeEndpointsInput) ([]*dms.Endpoint, error) {
+	var output []*dms.Endpoint
+
+	err := conn.DescribeEndpointsPagesWithContext(ctx, input, func(page *dms.DescribeEndpointsOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
+		}
+
+		for _, v := range page.Endpoints {
+			if v != nil {
+				output = append(output, v)
+			}
+		}
+
+		return !lastPage
+	})
+
+	if tfawserr.ErrCodeEquals(err, dms.ErrCodeResourceNotFoundFault) {
+		return nil, &retry.NotFoundError{
+			LastError:   err,
+			LastRequest: input,
+		}
+	}
+
+	if err != nil {
+		return nil, err
+	}
+
+	return output, nil
+}
+
+func statusEndpoint(ctx context.Context, conn *dms.DatabaseMigrationService, id string) retry.StateRefreshFunc {
+	return func() (interface{}, string, error) {
+		output, err := FindEndpointByID(ctx, conn, id)
+
+		if tfresource.NotFound(err) {
+			return nil, "", nil
+		}
+
+		if err != nil {
+			return nil, "", err
+		}
+
+		return output, aws.StringValue(output.Status), nil
+	}
+}
+
+func waitEndpointDeleted(ctx context.Context, conn *dms.DatabaseMigrationService, id string, timeout time.Duration) error {
+	stateConf := &retry.StateChangeConf{
+		Pending: []string{endpointStatusDeleting},
+		Target:  []string{},
+		Refresh: statusEndpoint(ctx, conn, id),
+		Timeout: timeout,
+	}
+
+	_, err := stateConf.WaitForStateContext(ctx)
+
+	return err
+}

internal/service/dms/endpoint_data_source.go

@@ -10,9 +10,8 @@ import (
 	"github.com/hashicorp/terraform-plugin-sdk/v2/diag"
 	"github.com/hashicorp/terraform-plugin-sdk/v2/helper/schema"
 	"github.com/hashicorp/terraform-provider-aws/internal/conns"
-	"github.com/hashicorp/terraform-provider-aws/internal/create"
+	"github.com/hashicorp/terraform-provider-aws/internal/errs/sdkdiag"
 	tftags "github.com/hashicorp/terraform-provider-aws/internal/tags"
-	"github.com/hashicorp/terraform-provider-aws/names"
 )
 
 // @SDKDataSource("aws_dms_endpoint")
@@ -31,12 +30,12 @@ func DataSourceEndpoint() *schema.Resource {
 			},
 			"elasticsearch_settings": {
 				Type:     schema.TypeList,
-				Optional: true,
+				Computed: true,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"endpoint_uri": {
 							Type:     schema.TypeString,
-							Required: true,
+							Computed: true,
 						},
 						"error_retry_duration": {
 							Type:     schema.TypeInt,
@@ -48,7 +47,7 @@ func DataSourceEndpoint() *schema.Resource {
 						},
 						"service_access_role_arn": {
 							Type:     schema.TypeString,
-							Required: true,
+							Computed: true,
 						},
 					},
 				},
@@ -71,16 +70,16 @@ func DataSourceEndpoint() *schema.Resource {
 			},
 			"extra_connection_attributes": {
 				Type:     schema.TypeString,
-				Optional: true,
+				Computed: true,
 			},
 			"kafka_settings": {
 				Type:     schema.TypeList,
-				Optional: true,
+				Computed: true,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"broker": {
 							Type:     schema.TypeString,
-							Required: true,
+							Computed: true,
 						},
 						"include_control_details": {
 							Type:     schema.TypeBool,
@@ -203,7 +202,7 @@ func DataSourceEndpoint() *schema.Resource {
 			},
 			"mongodb_settings": {
 				Type:     schema.TypeList,
-				Optional: true,
+				Computed: true,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"auth_mechanism": {
@@ -494,26 +493,23 @@ func DataSourceEndpoint() *schema.Resource {
 	}
 }
 
-const (
-	DSNameEndpoint = "Endpoint Data Source"
-)
-
 func dataSourceEndpointRead(ctx context.Context, d *schema.ResourceData, meta interface{}) diag.Diagnostics {
+	var diags diag.Diagnostics
 	conn := meta.(*conns.AWSClient).DMSConn(ctx)
 	defaultTagsConfig := meta.(*conns.AWSClient).DefaultTagsConfig
 	ignoreTagsConfig := meta.(*conns.AWSClient).IgnoreTagsConfig
 
 	endptID := d.Get("endpoint_id").(string)
-
 	out, err := FindEndpointByID(ctx, conn, endptID)
+
 	if err != nil {
-		create.DiagError(names.DMS, create.ErrActionReading, DSNameEndpoint, d.Id(), err)
+		return sdkdiag.AppendErrorf(diags, "reading DMS Endpoint (%s): %s", endptID, err)
 	}
 
 	d.SetId(aws.StringValue(out.EndpointIdentifier))
-
 	d.Set("endpoint_id", out.EndpointIdentifier)
-	d.Set("endpoint_arn", out.EndpointArn)
+	arn := aws.StringValue(out.EndpointArn)
+	d.Set("endpoint_arn", arn)
 	d.Set("endpoint_type", out.EndpointType)
 	d.Set("database_name", out.DatabaseName)
 	d.Set("engine_name", out.EngineName)
@@ -522,22 +518,21 @@ func dataSourceEndpointRead(ctx context.Context, d *schema.ResourceData, meta in
 	d.Set("ssl_mode", out.SslMode)
 	d.Set("username", out.Username)
 
-	err = resourceEndpointSetState(d, out)
-	if err != nil {
-		create.DiagError(names.DMS, create.ErrActionReading, DSNameEndpoint, d.Id(), err)
+	if err := resourceEndpointSetState(d, out); err != nil {
+		return sdkdiag.AppendFromErr(diags, err)
 	}
 
-	tags, err := listTags(ctx, conn, aws.StringValue(out.EndpointArn))
+	tags, err := listTags(ctx, conn, arn)
 	if err != nil {
-		return create.DiagError(names.DMS, create.ErrActionReading, DSNameEndpoint, d.Id(), err)
+		return sdkdiag.AppendErrorf(diags, "listing tags for DMS Endpoint (%s): %s", arn, err)
 	}
 
 	tags = tags.IgnoreAWS().IgnoreConfig(ignoreTagsConfig)
 
 	//lintignore:AWSR002
 	if err := d.Set("tags", tags.RemoveDefaultConfig(defaultTagsConfig).Map()); err != nil {
-		return create.DiagError(names.DMS, create.ErrActionSetting, DSNameEndpoint, d.Id(), err)
+		return sdkdiag.AppendErrorf(diags, "setting tags: %s", err)
 	}
 
-	return nil
+	return diags
 }

internal/service/dms/endpoint_data_source_test.go

@@ -14,10 +14,6 @@ import (
 )
 
 func TestAccDMSEndpointDataSource_basic(t *testing.T) {
-	if testing.Short() {
-		t.Skip("skipping long-running test in short mode")
-	}
-
 	ctx := acctest.Context(t)
 	rName := sdkacctest.RandomWithPrefix(acctest.ResourcePrefix)
 	resourceName := "aws_dms_endpoint.test"

internal/service/dms/find.go

@@ -13,40 +13,6 @@ import (
 	"github.com/hashicorp/terraform-provider-aws/internal/tfresource"
 )
 
-func FindEndpointByID(ctx context.Context, conn *dms.DatabaseMigrationService, id string) (*dms.Endpoint, error) {
-	input := &dms.DescribeEndpointsInput{
-		Filters: []*dms.Filter{
-			{
-				Name:   aws.String("endpoint-id"),
-				Values: aws.StringSlice([]string{id}),
-			},
-		},
-	}
-
-	output, err := conn.DescribeEndpointsWithContext(ctx, input)
-
-	if tfawserr.ErrCodeEquals(err, dms.ErrCodeResourceNotFoundFault) {
-		return nil, &retry.NotFoundError{
-			LastError:   err,
-			LastRequest: input,
-		}
-	}
-
-	if err != nil {
-		return nil, err
-	}
-
-	if output == nil || len(output.Endpoints) == 0 || output.Endpoints[0] == nil {
-		return nil, tfresource.NewEmptyResultError(input)
-	}
-
-	if count := len(output.Endpoints); count > 1 {
-		return nil, tfresource.NewTooManyResultsError(count, input)
-	}
-
-	return output.Endpoints[0], nil
-}
-
 func FindReplicationTaskByID(ctx context.Context, conn *dms.DatabaseMigrationService, id string) (*dms.ReplicationTask, error) {
 	input := &dms.DescribeReplicationTasksInput{
 		Filters: []*dms.Filter{