Update org_clean_test.go to delete stream instances & private endpoint services

Author: maastha

internal/testutil/clean/org_clean_test.go

@@ -11,11 +11,13 @@ import (
 	"testing"
 	"time"
 
+	"go.mongodb.org/atlas-sdk/v20250312001/admin"
+
+	"github.com/stretchr/testify/require"
+
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/constant"
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/common/dsschema"
 	"github.com/mongodb/terraform-provider-mongodbatlas/internal/testutil/acc"
-	"github.com/stretchr/testify/require"
-	"go.mongodb.org/atlas-sdk/v20250312002/admin"
 )
 
 const (
@@ -40,6 +42,8 @@ var (
 		"CANNOT_CLOSE_GROUP_ACTIVE_PEERING_CONNECTIONS",
 		"CANNOT_CLOSE_GROUP_ACTIVE_ATLAS_DATA_LAKES",
 		"CANNOT_CLOSE_GROUP_ACTIVE_ATLAS_DATA_FEDERATION_PRIVATE_ENDPOINTS",
+		"CANNOT_CLOSE_GROUP_ACTIVE_STREAMS_RESOURCE",
+		"CANNOT_CLOSE_GROUP_ACTIVE_ATLAS_PRIVATE_ENDPOINT_SERVICES",
 	}
 )
 
@@ -153,7 +157,7 @@ func findRetryErrorCode(err error) string {
 	return ""
 }
 func deleteProject(ctx context.Context, client *admin.APIClient, projectID string) error {
-	_, err := client.ProjectsApi.DeleteProject(ctx, projectID).Execute()
+	_, _, err := client.ProjectsApi.DeleteProject(ctx, projectID).Execute()
 	if err == nil || admin.IsErrorCode(err, "PROJECT_NOT_FOUND") {
 		return nil
 	}
@@ -187,6 +191,14 @@ func removeProjectResources(ctx context.Context, t *testing.T, dryRun bool, clie
 	if federatedDatabasesRemoved > 0 {
 		changes = append(changes, fmt.Sprintf("removed %d federated databases", federatedDatabasesRemoved))
 	}
+	streamInstancesRemoved := removeStreamInstances(ctx, t, dryRun, client, projectID)
+	if streamInstancesRemoved > 0 {
+		changes = append(changes, fmt.Sprintf("removed %d stream instances", streamInstancesRemoved))
+	}
+	privateEndpointServicesRemoved := removePrivateEndpointServices(ctx, t, dryRun, client, projectID)
+	if privateEndpointServicesRemoved > 0 {
+		changes = append(changes, fmt.Sprintf("removed %d private endpoint services", privateEndpointServicesRemoved))
+	}
 	return strings.Join(changes, ", ")
 }
 
@@ -296,13 +308,72 @@ func removeDataLakePipelines(ctx context.Context, t *testing.T, dryRun bool, cli
 		pipelineID := p.GetId()
 		t.Logf("delete pipeline %s", pipelineID)
 		if !dryRun {
-			_, err = client.DataLakePipelinesApi.DeletePipeline(ctx, projectID, pipelineID).Execute()
+			_, _, err = client.DataLakePipelinesApi.DeletePipeline(ctx, projectID, pipelineID).Execute()
 			require.NoError(t, err)
 		}
 	}
 	return len(datalakeResults)
 }
 
+func removeStreamInstances(ctx context.Context, t *testing.T, dryRun bool, client *admin.APIClient, projectID string) int {
+	t.Helper()
+	streamInstances, _, err := client.StreamsApi.ListStreamInstances(ctx, projectID).Execute()
+	require.NoError(t, err)
+
+	for _, instance := range *streamInstances.Results {
+		instanceName := *instance.Name
+		id := instance.GetId()
+		t.Logf("delete stream instance %s", id)
+
+		if !dryRun {
+			_, _, err = client.StreamsApi.DeleteStreamInstance(ctx, projectID, instanceName).Execute()
+			if err != nil && admin.IsErrorCode(err, "STREAM_TENANT_HAS_STREAM_PROCESSORS") {
+				t.Logf("stream instance %s has stream processors, attempting to delete", id)
+				streamProcessors, _, spErr := client.StreamsApi.ListStreamProcessors(ctx, projectID, instanceName).Execute()
+				require.NoError(t, spErr)
+
+				for _, processor := range *streamProcessors.Results {
+					t.Logf("delete stream processor %s", processor.Id)
+					_, err = client.StreamsApi.DeleteStreamProcessor(ctx, projectID, instanceName, processor.Name).Execute()
+					require.NoError(t, err)
+				}
+				t.Logf("retry delete stream instance %s after removing stream processors", id)
+				_, _, err = client.StreamsApi.DeleteStreamInstance(ctx, projectID, instanceName).Execute()
+				require.NoError(t, err)
+			} else {
+				require.NoError(t, err)
+			}
+		}
+	}
+	return len(*streamInstances.Results)
+}
+
+func removePrivateEndpointServices(ctx context.Context, t *testing.T, dryRun bool, client *admin.APIClient, projectID string) int {
+	t.Helper()
+	totalCount := 0
+	cloudProviders := []string{"AWS", "AZURE", "GCP"}
+
+	for _, provider := range cloudProviders {
+		endpointServices, _, err := client.PrivateEndpointServicesApi.ListPrivateEndpointServices(ctx, projectID, provider).Execute()
+		if err != nil {
+			t.Errorf("failed to list private endpoint services for %s: %v", provider, err)
+			continue
+		}
+
+		for _, service := range endpointServices {
+			id := service.GetId()
+			t.Logf("delete private endpoint service %s for provider %s", id, provider)
+			if !dryRun {
+				_, _, err := client.PrivateEndpointServicesApi.DeletePrivateEndpointService(ctx, projectID, provider, id).Execute()
+				require.NoError(t, err)
+			}
+		}
+		totalCount += len(endpointServices)
+	}
+
+	return totalCount
+}
+
 func removeFederatedDatabasePrivateEndpoints(ctx context.Context, t *testing.T, dryRun bool, client *admin.APIClient, projectID string) int {
 	t.Helper()
 	paginatedResults, _, err := client.DataFederationApi.ListDataFederationPrivateEndpoints(ctx, projectID).Execute()
@@ -312,7 +383,7 @@ func removeFederatedDatabasePrivateEndpoints(ctx context.Context, t *testing.T,
 		endpointID := f.GetEndpointId()
 		t.Logf("delete federated private endpoint %s", endpointID)
 		if !dryRun {
-			_, err = client.DataFederationApi.DeleteDataFederationPrivateEndpoint(ctx, projectID, endpointID).Execute()
+			_, _, err = client.DataFederationApi.DeleteDataFederationPrivateEndpoint(ctx, projectID, endpointID).Execute()
 			require.NoError(t, err)
 		}
 	}
@@ -331,7 +402,7 @@ func removeFederatedDatabases(ctx context.Context, t *testing.T, dryRun bool, cl
 		federatedName := f.GetName()
 		t.Logf("delete federated %s", federatedName)
 		if !dryRun {
-			_, err = client.DataFederationApi.DeleteFederatedDatabase(ctx, projectID, federatedName).Execute()
+			_, _, err = client.DataFederationApi.DeleteFederatedDatabase(ctx, projectID, federatedName).Execute()
 			require.NoError(t, err)
 		}
 	}