test: Cover direct stream processor start

Author: jwongmongodb

.changelog/4683.txt

@@ -111,8 +111,6 @@ func (r *streamProcessorRS) Create(ctx context.Context, req resource.CreateReque
 		if plan.Tier.ValueString() != "" {
 			startWithOptions.SetTier(plan.Tier.ValueString())
 		}
-		// Autoscaling is persisted through the preceding Create request. Do not repeat
-		// it in :startWith: that endpoint treats it as a new autoscaling modification.
 		_, err := connV2.StreamsAPI.StartStreamProcessorWith(ctx, projectID, workspaceOrInstanceName, processorName, startWithOptions).Execute()
 		if err != nil {
 			resp.Diagnostics.AddError(errorCreateStart, err.Error())
@@ -241,8 +239,6 @@ func (r *streamProcessorRS) Update(ctx context.Context, req resource.UpdateReque
 		if plan.Tier.ValueString() != "" {
 			startWithOptions.SetTier(plan.Tier.ValueString())
 		}
-		// Autoscaling is persisted through the preceding PATCH request. Do not repeat
-		// it in :startWith: that endpoint treats it as a new autoscaling modification.
 		_, err := r.Client.AtlasV2.StreamsAPI.StartStreamProcessorWith(ctx, projectID, workspaceOrInstanceName, processorName, startWithOptions).Execute()
 		if err != nil {
 			resp.Diagnostics.AddError("Error starting stream processor", err.Error())

internal/service/streamprocessor/resource.go

@@ -138,16 +138,27 @@ func TestAccStreamProcessor_withOptionsDLQAutoscaling(t *testing.T) {
 		CheckDestroy:             checkDestroyStreamProcessor,
 		Steps: []resource.TestStep{
 			{
-				// Autoscaling is persisted by Create before the provider starts the
-				// processor. The subsequent :startWith request does not repeat it.
 				Config: configWithOptionsDLQAutoscaling(t, projectID, workspaceName, clusterName, processorName, streamProcessorOptionsConfig{
 					includeDLQ:         true,
 					autoscalingMinTier: "SP10",
 					autoscalingMaxTier: "SP50",
-					state:              streamprocessor.StartedState,
+					state:              streamprocessor.CreatedState,
 				}),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.min_tier", "SP10"),
+					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.max_tier", "SP50"),
+					resource.TestCheckResourceAttrSet(resourceName, "effective_tier"),
+				),
+			},
+			{
+				Config: configWithOptionsDLQAutoscaling(t, projectID, workspaceName, clusterName, processorName, streamProcessorOptionsConfig{
+					includeDLQ:         true,
+					autoscalingMinTier: "SP10",
+					autoscalingMaxTier: "SP50",
+					state:              streamprocessor.StartedState,
+				}),
+				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttr(resourceName, "state", streamprocessor.StartedState),
 					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.min_tier", "SP10"),
 					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.max_tier", "SP50"),
@@ -185,6 +196,38 @@ func TestAccStreamProcessor_withOptionsDLQAutoscaling(t *testing.T) {
 		}})
 }
 
+func TestAccStreamProcessor_withOptionsDLQAutoscalingCreateStarted(t *testing.T) {
+	var (
+		projectID, workspaceName = acc.ProjectIDExecutionWithStreamInstance(t)
+		_, clusterName           = acc.ClusterNameExecution(t, false)
+		randomSuffix             = acctest.RandString(5)
+		processorName            = "new-processor-autoscaling-started-" + randomSuffix
+	)
+
+	resource.Test(t, resource.TestCase{
+		PreCheck:                 func() { acc.PreCheckBasic(t) },
+		ProtoV6ProviderFactories: acc.TestAccProviderV6Factories,
+		CheckDestroy:             checkDestroyStreamProcessor,
+		Steps: []resource.TestStep{
+			{
+				Config: configWithOptionsDLQAutoscaling(t, projectID, workspaceName, clusterName, processorName, streamProcessorOptionsConfig{
+					includeDLQ:         true,
+					autoscalingMinTier: "SP10",
+					autoscalingMaxTier: "SP50",
+					state:              streamprocessor.StartedState,
+				}),
+				Check: resource.ComposeAggregateTestCheckFunc(
+					resource.TestCheckResourceAttrSet(resourceName, "id"),
+					resource.TestCheckResourceAttr(resourceName, "state", streamprocessor.StartedState),
+					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.min_tier", "SP10"),
+					resource.TestCheckResourceAttr(resourceName, "options.autoscaling.max_tier", "SP50"),
+					resource.TestCheckResourceAttrSet(resourceName, "effective_tier"),
+				),
+			},
+			importStep(),
+		}})
+}
+
 func basicTestCase(t *testing.T) *resource.TestCase {
 	t.Helper()
 	var (