Add diff suppress for storage_config.bucket field (#11606) (#19552)

[upstream:a25e4a1b4f64d285504b7a46f4d200e7ee127116]

Signed-off-by: Modular Magician <[email protected]>

.changelog/11606.txt

@@ -0,0 +1,3 @@
+```release-note:enhancement
+composer: added diff suppress to `storage_config.bucket` field (users should be able to specify bucket name with or without "gs://" prefix). 
+```

google/services/composer/resource_composer_environment.go

@@ -924,10 +924,11 @@ func ResourceComposerEnvironment() *schema.Resource {
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
 						"bucket": {
-							Type:        schema.TypeString,
-							Required:    true,
-							ForceNew:    true,
-							Description: `Optional. Name of an existing Cloud Storage bucket to be used by the environment.`,
+							Type:             schema.TypeString,
+							Required:         true,
+							ForceNew:         true,
+							DiffSuppressFunc: gscBucketNameDiffSuppress,
+							Description:      `Optional. Name of an existing Cloud Storage bucket to be used by the environment.`,
 						},
 					},
 				},
@@ -2701,3 +2702,11 @@ func validateComposerInternalIpv4CidrBlock(v any, k string) (warns []string, err
 	}
 	return
 }
+
+func gscBucketNameDiffSuppress(_, old, new string, _ *schema.ResourceData) bool {
+	prefix := "gs://"
+	if prefix+old == new || old == prefix+new {
+		return true
+	}
+	return false
+}

google/services/composer/resource_composer_environment_test.go

@@ -1154,6 +1154,39 @@ func TestAccComposerEnvironment_customBucket(t *testing.T) {
 	})
 }
 
+func TestAccComposerEnvironment_customBucketWithUrl(t *testing.T) {
+	t.Parallel()
+
+	bucketName := fmt.Sprintf("%s-%d", testComposerBucketPrefix, acctest.RandInt(t))
+	envName := fmt.Sprintf("%s-%d", testComposerEnvironmentPrefix, acctest.RandInt(t))
+	network := fmt.Sprintf("%s-%d", testComposerNetworkPrefix, acctest.RandInt(t))
+	subnetwork := network + "-1"
+	acctest.VcrTest(t, resource.TestCase{
+		PreCheck:                 func() { acctest.AccTestPreCheck(t) },
+		ProtoV5ProviderFactories: acctest.ProtoV5ProviderFactories(t),
+		CheckDestroy:             testAccComposerEnvironmentDestroyProducer(t),
+		Steps: []resource.TestStep{
+			{
+				Config: testAccComposerEnvironment_customBucketWithUrl(bucketName, envName, network, subnetwork),
+			},
+			{
+				ResourceName:      "google_composer_environment.test",
+				ImportState:       true,
+				ImportStateVerify: true,
+			},
+			// This is a terrible clean-up step in order to get destroy to succeed,
+			// due to dangling firewall rules left by the Composer Environment blocking network deletion.
+			// TODO: Remove this check if firewall rules bug gets fixed by Composer.
+			{
+				PlanOnly:           true,
+				ExpectNonEmptyPlan: false,
+				Config:             testAccComposerEnvironment_customBucketWithUrl(bucketName, envName, network, subnetwork),
+				Check:              testAccCheckClearComposerEnvironmentFirewalls(t, network),
+			},
+		},
+	})
+}
+
 func testAccComposerEnvironment_customBucket(bucketName, envName, network, subnetwork string) string {
 	return fmt.Sprintf(`
 resource "google_storage_bucket" "test" {
@@ -1198,6 +1231,47 @@ resource "google_compute_subnetwork" "test" {
 `, bucketName, envName, network, subnetwork)
 }
 
+func testAccComposerEnvironment_customBucketWithUrl(bucketName, envName, network, subnetwork string) string {
+	return fmt.Sprintf(`
+resource "google_storage_bucket" "test" {
+  name   = "%s"
+  location = "us-central1"
+  force_destroy = true
+}
+
+resource "google_composer_environment" "test" {
+  name   = "%s"
+  region = "us-central1"
+  config {
+    node_config {
+      network        = google_compute_network.test.self_link
+      subnetwork     = google_compute_subnetwork.test.self_link
+    }
+    software_config {
+      image_version = "composer-2.9.3-airflow-2.9.1"
+    }
+  }
+  storage_config {
+		bucket = google_storage_bucket.test.url
+	}
+}
+
+// use a separate network to avoid conflicts with other tests running in parallel
+// that use the default network/subnet
+resource "google_compute_network" "test" {
+  name                    = "%s"
+  auto_create_subnetworks = false
+}
+
+resource "google_compute_subnetwork" "test" {
+  name          = "%s"
+  ip_cidr_range = "10.2.0.0/16"
+  region        = "us-central1"
+  network       = google_compute_network.test.self_link
+}
+`, bucketName, envName, network, subnetwork)
+}
+
 func testAccComposerEnvironment_basic(name, network, subnetwork string) string {
 	return fmt.Sprintf(`
 resource "google_composer_environment" "test" {