Repository: turbot/steampipe-plugin-aws (language: go)

Issue:
Add table aws_dms_replication_task
**References**
https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/dms_replication_task


Discussion / hints:

Fix patch:
diff --git a/aws/plugin.go b/aws/plugin.go
index ac7483243..8dbbbc35c 100644
--- a/aws/plugin.go
+++ b/aws/plugin.go
@@ -150,6 +150,7 @@ func Plugin(ctx context.Context) *plugin.Plugin {
 			"aws_dlm_lifecycle_policy":                                     tableAwsDLMLifecyclePolicy(ctx),
 			"aws_dms_certificate":                                          tableAwsDmsCertificate(ctx),
 			"aws_dms_replication_instance":                                 tableAwsDmsReplicationInstance(ctx),
+			"aws_dms_replication_task":                                     tableAwsDmsReplicationTask(ctx),
 			"aws_docdb_cluster":                                            tableAwsDocDBCluster(ctx),
 			"aws_docdb_cluster_instance":                                   tableAwsDocDBClusterInstance(ctx),
 			"aws_drs_job":                                                  tableAwsDRSJob(ctx),
diff --git a/aws/table_aws_dms_replication_task.go b/aws/table_aws_dms_replication_task.go
new file mode 100644
index 000000000..d71a45754
--- /dev/null
+++ b/aws/table_aws_dms_replication_task.go
@@ -0,0 +1,338 @@
+package aws
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/databasemigrationservice"
+	"github.com/aws/aws-sdk-go-v2/service/databasemigrationservice/types"
+
+	databasemigrationservicev1 "github.com/aws/aws-sdk-go/service/databasemigrationservice"
+
+	"github.com/turbot/steampipe-plugin-sdk/v5/grpc/proto"
+	"github.com/turbot/steampipe-plugin-sdk/v5/plugin"
+	"github.com/turbot/steampipe-plugin-sdk/v5/plugin/transform"
+)
+
+//// TABLE DEFINITION
+
+func tableAwsDmsReplicationTask(_ context.Context) *plugin.Table {
+	return &plugin.Table{
+		Name:        "aws_dms_replication_task",
+		Description: "AWS DMS Replication Task",
+		List: &plugin.ListConfig{
+			Hydrate: listDmsReplicationTasks,
+			// If the ARN provided as an input parameter refers to a resource that is unavailable in the specified region, the API throws an InvalidParameterValueException exception.
+			IgnoreConfig: &plugin.IgnoreConfig{
+				ShouldIgnoreErrorFunc: shouldIgnoreErrors([]string{"InvalidParameterValueException", "ResourceNotFoundFault"}),
+			},
+			Tags: map[string]string{"service": "dms", "action": "DescribeReplicationTasks"},
+			KeyColumns: []*plugin.KeyColumn{
+				{
+					Name:    "replication_task_identifier",
+					Require: plugin.Optional,
+				},
+				{
+					Name:    "arn",
+					Require: plugin.Optional,
+				},
+				{
+					Name:    "replication_instance_arn",
+					Require: plugin.Optional,
+				},
+				{
+					Name:    "target_endpoint_arn",
+					Require: plugin.Optional,
+				},
+				{
+					Name:    "source_endpoint_arn",
+					Require: plugin.Optional,
+				},
+				{
+					Name:    "migration_type",
+					Require: plugin.Optional,
+				},
+			},
+		},
+		HydrateConfig: []plugin.HydrateConfig{
+			{
+				Func: getDmsReplicationTaskTags,
+				Tags: map[string]string{"service": "dms", "action": "ListTagsForResource"},
+			},
+		},
+		GetMatrixItemFunc: SupportedRegionMatrix(databasemigrationservicev1.EndpointsID),
+		Columns: awsRegionalColumns([]*plugin.Column{
+			{
+				Name:        "replication_task_identifier",
+				Description: "The user-assigned replication task identifier or name.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "arn",
+				Description: "The Amazon Resource Name (ARN) of the replication task.",
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("ReplicationTaskArn"),
+			},
+			{
+				Name:        "cdc_start_position",
+				Description: "Indicates when you want a change data capture (CDC) operation to start.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "cdc_stop_position",
+				Description: "Indicates when you want a change data capture (CDC) operation to stop.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "last_failure_message",
+				Description: "The last error (failure) message generated for the replication task.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "migration_type",
+				Description: "The type of migration.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "recovery_checkpoint",
+				Description: "Indicates the last checkpoint that occurred during a change data capture (CDC) operation.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "replication_instance_arn",
+				Description: "The Amazon Resource Name (ARN) of the replication instance.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "replication_task_creation_date",
+				Description: "The date the replication task was created.",
+				Type:        proto.ColumnType_TIMESTAMP,
+			},
+			{
+				Name:        "replication_task_start_date",
+				Description: "The date the replication task is scheduled to start.",
+				Type:        proto.ColumnType_TIMESTAMP,
+			},
+			{
+				Name:        "source_endpoint_arn",
+				Description: "The Amazon Resource Name (ARN) that uniquely identifies the endpoint.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "status",
+				Description: "The status of the replication task.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "stop_reason",
+				Description: "The reason the replication task was stopped.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "table_mappings",
+				Description: "Table mappings specified in the task.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "target_endpoint_arn",
+				Description: "The ARN that uniquely identifies the endpoint.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "target_replication_instance_arn",
+				Description: "The ARN of the replication instance to which this task is moved in response to running the MoveReplicationTask operation.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "task_data",
+				Description: "Supplemental information that the task requires to migrate the data for certain source and target endpoints.",
+				Type:        proto.ColumnType_STRING,
+			},
+			{
+				Name:        "replication_task_settings",
+				Description: "The settings for the replication task.",
+				Type:        proto.ColumnType_JSON,
+			},
+			{
+				Name:        "replication_task_stats",
+				Description: "The statistics for the task, including elapsed time, tables loaded, and table errors.",
+				Type:        proto.ColumnType_JSON,
+			},
+			{
+				Name:        "tags_src",
+				Description: "A list of tags currently associated with the replication instance.",
+				Type:        proto.ColumnType_JSON,
+				Hydrate:     getDmsReplicationTaskTags,
+				Transform:   transform.FromField("TagList"),
+			},
+
+			// Steampipe Standard Columns
+			{
+				Name:        "title",
+				Description: resourceInterfaceDescription("title"),
+				Type:        proto.ColumnType_STRING,
+				Transform:   transform.FromField("ReplicationTaskIdentifier"),
+			},
+			{
+				Name:        "tags",
+				Description: resourceInterfaceDescription("tags"),
+				Type:        proto.ColumnType_JSON,
+				Hydrate:     getDmsReplicationTaskTags,
+				Transform:   transform.From(dmsReplicationTaskTagListToTagsMap),
+			},
+			{
+				Name:        "akas",
+				Description: resourceInterfaceDescription("akas"),
+				Type:        proto.ColumnType_JSON,
+				Transform:   transform.FromField("ReplicationTaskArn").Transform(transform.EnsureStringArray),
+			},
+		}),
+	}
+}
+
+//// LIST FUNCTION
+
+func listDmsReplicationTasks(ctx context.Context, d *plugin.QueryData, _ *plugin.HydrateData) (interface{}, error) {
+	// Create Session
+	svc, err := DatabaseMigrationClient(ctx, d)
+	if err != nil {
+		plugin.Logger(ctx).Error("aws_dms_replication_task.listDmsReplicationTasks", "connection_error", err)
+		return nil, err
+	}
+
+	// Limiting the results
+	maxLimit := int32(100)
+	if d.QueryContext.Limit != nil {
+		limit := int32(*d.QueryContext.Limit)
+		if limit < maxLimit {
+			if limit < 20 {
+				maxLimit = 20
+			} else {
+				maxLimit = limit
+			}
+		}
+	}
+
+	// Build the params
+	input := &databasemigrationservice.DescribeReplicationTasksInput{
+		MaxRecords: aws.Int32(maxLimit),
+	}
+
+	var filter []types.Filter
+
+	// Additonal Filter
+	equalQuals := d.EqualsQuals
+	if equalQuals["replication_task_identifier"] != nil {
+		paramFilter := types.Filter{
+			Name:   aws.String("replication-task-id"),
+			Values: []string{equalQuals["replication_task_identifier"].GetStringValue()},
+		}
+		filter = append(filter, paramFilter)
+	}
+	if equalQuals["arn"] != nil {
+		paramFilter := types.Filter{
+			Name:   aws.String("replication-task-arn"),
+			Values: []string{equalQuals["arn"].GetStringValue()},
+		}
+		filter = append(filter, paramFilter)
+	}
+	if equalQuals["replication_instance_arn"] != nil {
+		paramFilter := types.Filter{
+			Name:   aws.String("replication-instance-arn"),
+			Values: []string{equalQuals["replication_instance_arn"].GetStringValue()},
+		}
+		filter = append(filter, paramFilter)
+	}
+	if equalQuals["migration_type"] != nil {
+		paramFilter := types.Filter{
+			Name:   aws.String("migration-type"),
+			Values: []string{equalQuals["migration_type"].GetStringValue()},
+		}
+		filter = append(filter, paramFilter)
+	}
+	if equalQuals["target_endpoint_arn"] != nil || equalQuals["source_endpoint_arn"] != nil {
+		paramFilter := types.Filter{
+			Name:   aws.String("endpoint-arn"),
+		}
+		values := []string{}
+
+		if equalQuals["target_endpoint_arn"].GetStringValue() != "" {
+			values = append(values, equalQuals["target_endpoint_arn"].GetStringValue())
+		}
+		if equalQuals["source_endpoint_arn"].GetStringValue() != "" {
+			values = append(values, equalQuals["source_endpoint_arn"].GetStringValue())
+		}
+		paramFilter.Values = values
+		filter = append(filter, paramFilter)
+	}
+	input.Filters = filter
+
+	paginator := databasemigrationservice.NewDescribeReplicationTasksPaginator(svc, input, func(o *databasemigrationservice.DescribeReplicationTasksPaginatorOptions) {
+		o.Limit = maxLimit
+		o.StopOnDuplicateToken = true
+	})
+
+	// List call
+	for paginator.HasMorePages() {
+		// apply rate limiting
+		d.WaitForListRateLimit(ctx)
+
+		output, err := paginator.NextPage(ctx)
+		if err != nil {
+			plugin.Logger(ctx).Error("aws_dms_replication_task.listDmsReplicationTasks", "api_error", err)
+			return nil, err
+		}
+
+		for _, items := range output.ReplicationTasks {
+			d.StreamListItem(ctx, items)
+
+			// Context can be cancelled due to manual cancellation or the limit has been hit
+			if d.RowsRemaining(ctx) == 0 {
+				return nil, nil
+			}
+		}
+	}
+	return nil, err
+}
+
+//// HYDRATE FUNCTIONS
+
+func getDmsReplicationTaskTags(ctx context.Context, d *plugin.QueryData, h *plugin.HydrateData) (interface{}, error) {
+
+	taskArn := h.Item.(types.ReplicationTask).ReplicationTaskArn
+
+	// Create service
+	svc, err := DatabaseMigrationClient(ctx, d)
+	if err != nil {
+		plugin.Logger(ctx).Error("aws_dms_replication_task.getDmsReplicationTaskTags", "connection_error", err)
+		return nil, err
+	}
+
+	params := &databasemigrationservice.ListTagsForResourceInput{
+		ResourceArn: taskArn,
+	}
+
+	replicationInstanceTags, err := svc.ListTagsForResource(ctx, params)
+	if err != nil {
+		plugin.Logger(ctx).Error("aws_dms_replication_task.getDmsReplicationTaskTags", "api_error", err)
+		return nil, err
+	}
+
+	return replicationInstanceTags, nil
+}
+
+//// TRANSFORM FUNCTIONS
+
+func dmsReplicationTaskTagListToTagsMap(_ context.Context, d *transform.TransformData) (interface{}, error) {
+	data := d.HydrateItem.(*databasemigrationservice.ListTagsForResourceOutput)
+
+	// Mapping the resource tags inside turbotTags
+	if data.TagList != nil {
+		turbotTagsMap := map[string]string{}
+		for _, i := range data.TagList {
+			turbotTagsMap[*i.Key] = *i.Value
+		}
+		return turbotTagsMap, nil
+	}
+	return nil, nil
+}
diff --git a/docs/tables/aws_dms_replication_task.md b/docs/tables/aws_dms_replication_task.md
new file mode 100644
index 000000000..f668f0d4b
--- /dev/null
+++ b/docs/tables/aws_dms_replication_task.md
@@ -0,0 +1,206 @@
+---
+title: "Steampipe Table: aws_dms_replication_task - Query AWS DMS Replication Tasks using SQL"
+description: "Enables users to query AWS DMS Replication Tasks to retrieve detailed information on data migration activities between source and target databases."
+---
+
+# Table: aws_dms_replication_task - Query AWS DMS Replication Tasks using SQL
+
+AWS Database Migration Service (DMS) Replication Tasks play a critical role in managing data migrations between source and target databases. These tasks facilitate the entire migration process, supporting various migration types, including full load migrations, ongoing replication to synchronize source and target databases and change data capture (CDC) for applying data modifications.
+
+The `aws_dms_replication_task` table in Steampipe allows for in-depth analysis of replication tasks, providing details such as task identifiers, status, migration types, settings, and endpoint ARNs. This table proves essential for database administrators and DevOps engineers overseeing database migrations, offering comprehensive insights into each task's configuration, progress, and performance.
+
+## Examples
+
+### Basic Info
+Query to fetch basic details about DMS replication tasks.
+
+```sql+postgresql
+select
+  replication_task_identifier,
+  arn,
+  migration_type,
+  status,
+  replication_task_creation_date
+from
+  aws_dms_replication_task;
+```
+
+```sql+sqlite
+select
+  replication_task_identifier,
+  arn,
+  migration_type,
+  status,
+  replication_task_creation_date
+from
+  aws_dms_replication_task;
+```
+
+### Tasks with specific migration types
+List replication tasks by a specific migration type, such as 'full-load'.
+
+```sql+postgresql
+select
+  replication_task_identifier,
+  migration_type,
+  status
+from
+  aws_dms_replication_task
+where
+  migration_type = 'full-load';
+```
+
+```sql+sqlite
+select
+  replication_task_identifier,
+  migration_type,
+  status
+from
+  aws_dms_replication_task
+where
+  migration_type = 'full-load';
+```
+
+### Replication tasks with failures
+Identify replication tasks that have failed, focusing on the last failure message.
+
+```sql+postgresql
+select
+  replication_task_identifier,
+  status,
+  last_failure_message
+from
+  aws_dms_replication_task
+where
+  status = 'failed';
+```
+
+```sql+sqlite
+select
+  replication_task_identifier,
+  status,
+  last_failure_message
+from
+  aws_dms_replication_task
+where
+  status = 'failed';
+```
+
+### Task performance statistics
+Examine detailed performance statistics of replication tasks.
+
+```sql+postgresql
+select
+  replication_task_identifier,
+  status,
+  replication_task_stats -> 'ElapsedTimeMillis' as elapsed_time_millis,
+  replication_task_stats -> 'FreshStartDate' as fresh_start_date,
+  replication_task_stats -> 'FullLoadFinishDate' as full_load_finish_date,
+  replication_task_stats -> 'FullLoadProgressPercent' as full_load_progress_percent,
+  replication_task_stats -> 'FullLoadStartDate' as full_load_start_date,
+  replication_task_stats -> 'StartDate' as start_date,
+  replication_task_stats -> 'StopDate' as stop_date,
+  replication_task_stats -> 'TablesErrored' as tables_errored,
+  replication_task_stats -> 'TablesLoaded' as tables_loaded,
+  replication_task_stats -> 'TablesLoading' as tables_loading,
+  replication_task_stats -> 'TablesQueued' as tables_queued
+from
+  aws_dms_replication_task;
+```
+
+```sql+sqlite
+select
+  replication_task_identifier,
+  status,
+  json_extract(replication_task_stats, '$.ElapsedTimeMillis') as elapsed_time_millis,
+  json_extract(replication_task_stats, '$.FreshStartDate') as fresh_start_date,
+  json_extract(replication_task_stats, '$.FullLoadFinishDate') as full_load_finish_date,
+  json_extract(replication_task_stats, '$.FullLoadProgressPercent') as full_load_progress_percent,
+  json_extract(replication_task_stats, '$.FullLoadStartDate') as full_load_start_date,
+  json_extract(replication_task_stats, '$.StartDate') as start_date,
+  json_extract(replication_task_stats, '$.StopDate') as stop_date,
+  json_extract(replication_task_stats, '$.TablesErrored') as tables_errored,
+  json_extract(replication_task_stats, '$.TablesLoaded') as tables_loaded,
+  json_extract(replication_task_stats, '$.TablesLoading') as tables_loading,
+  json_extract(replication_task_stats, '$.TablesQueued') as tables_queued
+from
+  aws_dms_replication_task;
+```
+
+### Get replication instance details
+Retrieve replication instance details for the tasks.
+
+```sql+postgresql
+select
+  t.replication_task_identifier,
+  t.arn as task_arn,
+  i.replication_instance_class,
+  i.engine_version,
+  i.publicly_accessible,
+  i.dns_name_servers
+from
+  aws_dms_replication_task t
+join aws_dms_replication_instance i on t.replication_instance_arn = i.arn;
+```
+
+```sql+sqlite
+select
+  t.replication_task_identifier,
+  t.arn as task_arn,
+  i.replication_instance_class,
+  i.engine_version,
+  i.publicly_accessible,
+  i.dns_name_servers
+from
+  aws_dms_replication_task as t
+join
+  aws_dms_replication_instance as i on t.replication_instance_arn = i.arn;
+```
+
+### List source endpoint tasks
+Query to list tasks associated with source endpoints.
+
+```sql+postgresql
+select
+  replication_task_identifier,
+  source_endpoint_arn,
+  status
+from
+  aws_dms_replication_task
+where
+  endpoint_type = 'source';
+```
+
+```sql+sqlite
+select
+  replication_task_identifier,
+  source_endpoint_arn,
+  status
+from
+  aws_dms_replication_task
+where
+  endpoint_type = 'source';
+```
+
+### Endpoint type count
+Count tasks by endpoint type (source or target).
+
+```sql+postgresql
+select
+  endpoint_type,
+  count(*) as task_count
+from
+  aws_dms_replication_task
+group by
+  endpoint_type;
+```
+
+```sql+sqlite
+select
+  endpoint_type,
+  count(*) as task_count
+from
+  aws_dms_replication_task
+group by
+  endpoint_type;
+```


Test patch:
diff --git a/aws-test/tests/aws_dms_replication_task/dependencies.txt b/aws-test/tests/aws_dms_replication_task/dependencies.txt
new file mode 100644
index 000000000..e69de29bb
diff --git a/aws-test/tests/aws_dms_replication_task/test-get-expected.json b/aws-test/tests/aws_dms_replication_task/test-get-expected.json
new file mode 100644
index 000000000..fa92d1512
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-get-expected.json
@@ -0,0 +1,24 @@
+[
+  {
+    "account_id": "{{ output.account_id.value }}",
+    "akas": ["{{ output.resource_aka.value }}"],
+    "arn": "{{ output.resource_aka.value }}",
+    "migration_type": "full-load",
+    "partition": "{{ output.aws_partition.value }}",
+    "region": "{{ output.aws_region.value }}",
+    "replication_instance_arn": "{{ output.replication_instance_arn.value }}",
+    "replication_task_identifier": "{{ resourceName }}",
+    "tags": {
+      "Name": "test"
+    },
+    "tags_src": [
+      {
+        "Key": "Name",
+        "ResourceArn": null,
+        "Value": "test"
+      }
+    ],
+    "target_endpoint_arn": "{{ output.target_endpoint_arn.value }}",
+    "title": "{{ resourceName }}"
+  }
+]
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-get-query.sql b/aws-test/tests/aws_dms_replication_task/test-get-query.sql
new file mode 100644
index 000000000..ca1572182
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-get-query.sql
@@ -0,0 +1,3 @@
+select replication_task_identifier, arn, migration_type, replication_instance_arn, target_endpoint_arn, tags_src, title, tags, akas, partition, region, account_id
+from aws_dms_replication_task
+where arn = '{{ output.resource_aka.value }}';
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-list-expected.json b/aws-test/tests/aws_dms_replication_task/test-list-expected.json
new file mode 100644
index 000000000..318ac91d0
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-list-expected.json
@@ -0,0 +1,8 @@
+[
+  {
+    "arn": "{{ output.resource_aka.value }}",
+    "region": "{{ output.aws_region.value }}",
+    "replication_task_identifier": "{{ resourceName }}",
+    "title": "{{ resourceName }}"
+  }
+]
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-list-query.sql b/aws-test/tests/aws_dms_replication_task/test-list-query.sql
new file mode 100644
index 000000000..b0d98c426
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-list-query.sql
@@ -0,0 +1,3 @@
+select replication_task_identifier, arn, title, region
+from aws_dms_replication_task
+where akas::text = '["{{ output.resource_aka.value }}"]';
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-notfound-expected.json b/aws-test/tests/aws_dms_replication_task/test-notfound-expected.json
new file mode 100644
index 000000000..0637a088a
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-notfound-expected.json
@@ -0,0 +1,1 @@
+[]
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-notfound-query.sql b/aws-test/tests/aws_dms_replication_task/test-notfound-query.sql
new file mode 100644
index 000000000..29a807f24
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-notfound-query.sql
@@ -0,0 +1,3 @@
+select *
+from aws_dms_replication_task
+where arn = '{{ output.resource_aka.value }}1p000';
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-turbot-expected.json b/aws-test/tests/aws_dms_replication_task/test-turbot-expected.json
new file mode 100644
index 000000000..c19565363
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-turbot-expected.json
@@ -0,0 +1,9 @@
+[
+  {
+    "akas": ["{{ output.resource_aka.value }}"],
+    "tags": {
+      "Name": "test"
+    },
+    "title": "{{ resourceName }}"
+  }
+]
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/test-turbot-query.sql b/aws-test/tests/aws_dms_replication_task/test-turbot-query.sql
new file mode 100644
index 000000000..1a7b17f0a
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/test-turbot-query.sql
@@ -0,0 +1,3 @@
+select title, tags, akas
+from aws_dms_replication_task
+where arn = '{{ output.resource_aka.value }}';
\ No newline at end of file
diff --git a/aws-test/tests/aws_dms_replication_task/variables.json b/aws-test/tests/aws_dms_replication_task/variables.json
new file mode 100644
index 000000000..0967ef424
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/variables.json
@@ -0,0 +1,1 @@
+{}
diff --git a/aws-test/tests/aws_dms_replication_task/variables.tf b/aws-test/tests/aws_dms_replication_task/variables.tf
new file mode 100644
index 000000000..4b8565d51
--- /dev/null
+++ b/aws-test/tests/aws_dms_replication_task/variables.tf
@@ -0,0 +1,178 @@
+
+variable "resource_name" {
+  type        = string
+  default     = "turbottest-20200125"
+  description = "Name of the resource used throughout the test."
+}
+
+variable "aws_profile" {
+  type        = string
+  default     = "default"
+  description = "AWS credentials profile used for the test. Default is to use the default profile."
+}
+
+variable "aws_region" {
+  type        = string
+  default     = "us-east-1"
+  description = "AWS region used for the test. Does not work with default region in config, so must be defined here."
+}
+
+variable "aws_region_alternate" {
+  type        = string
+  default     = "us-east-2"
+  description = "Alternate AWS region used for tests that require two regions (e.g. DynamoDB global tables)."
+}
+
+provider "aws" {
+  profile = var.aws_profile
+  region  = var.aws_region
+}
+
+provider "aws" {
+  alias   = "alternate"
+  profile = var.aws_profile
+  region  = var.aws_region_alternate
+}
+
+data "aws_partition" "current" {}
+data "aws_caller_identity" "current" {}
+data "aws_region" "primary" {}
+data "aws_region" "alternate" {
+  provider = aws.alternate
+}
+
+data "null_data_source" "resource" {
+  inputs = {
+    scope = "arn:${data.aws_partition.current.partition}:::${data.aws_caller_identity.current.account_id}"
+  }
+}
+
+resource "aws_vpc" "main" {
+  cidr_block = "10.0.0.0/16"
+  tags = {
+    name = var.resource_name
+  }
+}
+
+resource "aws_subnet" "sub1" {
+  vpc_id            = aws_vpc.main.id
+  availability_zone = "us-east-1a"
+  cidr_block        = "10.0.0.0/17"
+  tags = {
+    Name = var.resource_name
+  }
+}
+resource "aws_subnet" "sub2" {
+  vpc_id            = aws_vpc.main.id
+  availability_zone = "us-east-1b"
+  cidr_block        = "10.0.128.0/17"
+  tags = {
+    Name = var.resource_name
+  }
+}
+
+# Create a new replication subnet group
+resource "aws_dms_replication_subnet_group" "named_test_resource" {
+  replication_subnet_group_description = "Example replication subnet group"
+  replication_subnet_group_id          = var.resource_name
+
+  subnet_ids = [
+    aws_subnet.sub1.id,
+    aws_subnet.sub2.id,
+  ]
+}
+
+resource "aws_dms_replication_instance" "named_test_resource" {
+  depends_on                   = [aws_dms_replication_subnet_group.named_test_resource]
+  allocated_storage            = 5
+  apply_immediately            = true
+  auto_minor_version_upgrade   = true
+  availability_zone            = "${var.aws_region}a"
+  multi_az                     = false
+  preferred_maintenance_window = "sun:10:30-sun:14:30"
+  publicly_accessible          = false
+  replication_instance_class   = "dms.t2.micro"
+  replication_instance_id      = var.resource_name
+  replication_subnet_group_id  = aws_dms_replication_subnet_group.named_test_resource.id
+  tags = {
+    foo = "bar"
+  }
+}
+
+resource "aws_dms_endpoint" "named_test_resource_target" {
+  depends_on                  = [aws_dms_replication_instance.named_test_resource]
+  database_name               = "test"
+  endpoint_id                 = var.resource_name
+  endpoint_type               = "target"
+  engine_name                 = "aurora"
+  extra_connection_attributes = ""
+  password                    = "test"
+  port                        = 3306
+  server_name                 = "test"
+  ssl_mode                    = "none"
+  username                    = "test"
+}
+resource "aws_dms_endpoint" "named_test_resource_source" {
+  depends_on                  = [aws_dms_replication_instance.named_test_resource]
+  database_name               = "test"
+  endpoint_id                 = "${var.resource_name}source"
+  endpoint_type               = "source"
+  engine_name                 = "aurora"
+  extra_connection_attributes = ""
+  password                    = "test"
+  port                        = 3305
+  server_name                 = "test"
+  ssl_mode                    = "none"
+  username                    = "test"
+}
+
+
+# Create a new replication task
+resource "aws_dms_replication_task" "named_test_resource" {
+  depends_on               = [aws_dms_replication_instance.named_test_resource, aws_dms_endpoint.named_test_resource_source, aws_dms_endpoint.named_test_resource_target]
+  cdc_start_time           = "1993-05-21T05:50:00Z"
+  migration_type           = "full-load"
+  replication_instance_arn = aws_dms_replication_instance.named_test_resource.replication_instance_arn
+  replication_task_id      = var.resource_name
+  source_endpoint_arn      = aws_dms_endpoint.named_test_resource_source.endpoint_arn
+  table_mappings           = "{\"rules\":[{\"rule-type\":\"selection\",\"rule-id\":\"1\",\"rule-name\":\"1\",\"object-locator\":{\"schema-name\":\"%\",\"table-name\":\"%\"},\"rule-action\":\"include\"}]}"
+
+  tags = {
+    Name = "test"
+  }
+
+  target_endpoint_arn = aws_dms_endpoint.named_test_resource_target.endpoint_arn
+}
+
+output "account_id" {
+  value = data.aws_caller_identity.current.account_id
+}
+
+output "aws_region" {
+  value = data.aws_region.primary.name
+}
+
+output "aws_partition" {
+  value = data.aws_partition.current.partition
+}
+
+output "resource_aka" {
+  value = aws_dms_replication_task.named_test_resource.replication_task_arn
+}
+
+output "target_endpoint_arn" {
+  value = aws_dms_endpoint.named_test_resource_target.endpoint_arn
+}
+
+output "replication_instance_arn" {
+  value = aws_dms_replication_instance.named_test_resource.replication_instance_arn
+}
+
+output "status" {
+  value = aws_dms_replication_task.named_test_resource.status
+}
+
+
+output "resource_name" {
+  value = var.resource_name
+}