From 5fe31d4861b40597fd72f6fb87f8c7427c06450b Mon Sep 17 00:00:00 2001
From: dstrates <99311378+dstrates@users.noreply.github.com>
Date: Tue, 27 Aug 2024 10:11:58 +1000
Subject: [PATCH] fix: resolve schema normalization (#43)

---
 README.md                                     |  6 +-
 .../schemaregistry_schema/resource.tf         | 42 ++++++++++++--
 internal/provider/data_source_schema.go       | 16 +-----
 internal/provider/data_source_schema_test.go  | 22 +++++--
 internal/provider/provider_test.go            |  3 +-
 internal/provider/resource_schema.go          | 54 +++---------------
 internal/provider/resource_schema_test.go     | 57 +++++++++++++------
 internal/provider/utils.go                    | 51 +++++++----------
 8 files changed, 129 insertions(+), 122 deletions(-)

diff --git a/README.md b/README.md
index 7c40104..205ccd6 100644
--- a/README.md
+++ b/README.md
@@ -24,9 +24,7 @@ If you want to build the provider from source, follow these steps:
 ## Testing the Provider
 
 The acceptance tests rely on [Testcontainers for Go (Redpanda)](https://golang.testcontainers.org/modules/redpanda/) to
-provide a Schema Registry API.
-
-This has some limitations:
+provide a Schema Registry API. This has some limitations:
 
 - Redpanda only supports `AVRO` and `PROTOBUF` encoding, cannot test `JSON` schemas
   [[1]](https://github.com/redpanda-data/redpanda/issues/6220)
@@ -42,7 +40,7 @@ terraform {
   required_providers {
     schemaregistry = {
       source = "cultureamp/schemaregistry"
-      version = "1.1.0"
+      version = "1.2.1"
     }
   }
 }
diff --git a/examples/resources/schemaregistry_schema/resource.tf b/examples/resources/schemaregistry_schema/resource.tf
index 1f47aad..0bcfdb6 100644
--- a/examples/resources/schemaregistry_schema/resource.tf
+++ b/examples/resources/schemaregistry_schema/resource.tf
@@ -1,7 +1,41 @@
-resource "schemaregistry_schema" "example" {
-  subject             = "example-subject"
-  schema              = "{\"type\":\"record\",\"name\":\"Test\",\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}"
+resource "schemaregistry_schema" "ref_01" {
+  subject             = "%s"
   schema_type         = "AVRO"
-  compatibility_level = "FORWARD_TRANSITIVE"
+  compatibility_level = "NONE"
+  schema              = <<EOF
+{
+  "type": "record",
+  "name": "Test",
+  "fields": [
+    {
+      "name": "f1",
+      "type": "string"
+    }
+  ]
+}
+EOF
+}
+
+resource "schemaregistry_schema" "example_01" {
+  subject             = "%s"
+  schema_type         = "AVRO"
+  compatibility_level = "NONE"
   hard_delete         = true
+  schema = jsonencode({
+    "type" : "record",
+    "name" : "Example",
+    "fields" : [
+      {
+        "name" : "f1",
+        "type" : "string"
+      }
+    ]
+  })
+  references = [
+    {
+      name    = "TestRef01"
+      subject = schemaregistry_schema.ref_01.subject
+      version = schemaregistry_schema.ref_01.version
+    },
+  ]
 }
diff --git a/internal/provider/data_source_schema.go b/internal/provider/data_source_schema.go
index b37f4d8..2c95e25 100644
--- a/internal/provider/data_source_schema.go
+++ b/internal/provider/data_source_schema.go
@@ -8,7 +8,6 @@ import (
 	"github.com/hashicorp/terraform-plugin-framework-validators/stringvalidator"
 	"github.com/hashicorp/terraform-plugin-framework/datasource"
 	"github.com/hashicorp/terraform-plugin-framework/datasource/schema"
-	"github.com/hashicorp/terraform-plugin-framework/diag"
 	"github.com/hashicorp/terraform-plugin-framework/schema/validator"
 	"github.com/hashicorp/terraform-plugin-framework/types"
 	"github.com/riferrei/srclient"
@@ -182,7 +181,7 @@ func (d *schemaDataSource) Read(ctx context.Context, req datasource.ReadRequest,
 	}
 
 	// Map response body to schema data source model
-	outputs := d.mapSchemaToOutputs(subject, schema, compatibilityLevel, &resp.Diagnostics)
+	outputs := d.mapSchemaToOutputs(subject, schema, compatibilityLevel)
 
 	// Set state to fully populated data
 	diags = resp.State.Set(ctx, outputs)
@@ -207,21 +206,12 @@ func (d *schemaDataSource) fetchCompatibilityLevel(subject string) (*srclient.Co
 
 // mapSchemaToOutputs maps the schema and compatibility level to the schema data source model.
 func (d *schemaDataSource) mapSchemaToOutputs(subject string, schema *srclient.Schema,
-	compatibilityLevel *srclient.CompatibilityLevel, diagnostics *diag.Diagnostics) schemaDataSourceModel {
-	// Normalize the schema string
-	normalizedSchema, err := NormalizeJSON(schema.Schema(), diagnostics)
-	if err != nil {
-		diagnostics.AddError(
-			"Normalization Error",
-			fmt.Sprintf("Could not normalize schema: %s", err),
-		)
-		return schemaDataSourceModel{}
-	}
+	compatibilityLevel *srclient.CompatibilityLevel) schemaDataSourceModel {
 
 	return schemaDataSourceModel{
 		ID:                 types.StringValue(subject),
 		Subject:            types.StringValue(subject),
-		Schema:             jsontypes.NewNormalizedValue(normalizedSchema),
+		Schema:             jsontypes.NewNormalizedValue(schema.Schema()),
 		SchemaID:           types.Int64Value(int64(schema.ID())),
 		SchemaType:         types.StringValue(FromSchemaType(schema.SchemaType())),
 		Version:            types.Int64Value(int64(schema.Version())),
diff --git a/internal/provider/data_source_schema_test.go b/internal/provider/data_source_schema_test.go
index bd42406..1ba4f16 100644
--- a/internal/provider/data_source_schema_test.go
+++ b/internal/provider/data_source_schema_test.go
@@ -19,9 +19,11 @@ func TestAccSchemaDataSource_basic(t *testing.T) {
 				Config: testAccSchemaDataSourceConfig_single(subjectName),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttr(datasourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(datasourceName, "schema", NormalizedJSON(initialSchema)),
 					resource.TestCheckResourceAttr(datasourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(datasourceName, "compatibility_level", "NONE"),
+					resource.TestCheckResourceAttrWith(datasourceName, "schema", func(state string) error {
+						return ValidateSchemaString(initialSchema, state)
+					}),
 				),
 			},
 		},
@@ -41,9 +43,11 @@ func TestAccSchemaDataSource_multipleVersions(t *testing.T) {
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrSet(datasourceName, "id"),
 					resource.TestCheckResourceAttr(datasourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(datasourceName, "schema", NormalizedJSON(initialSchema)),
 					resource.TestCheckResourceAttr(datasourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(datasourceName, "compatibility_level", "NONE"),
+					resource.TestCheckResourceAttrWith(datasourceName, "schema", func(state string) error {
+						return ValidateSchemaString(initialSchema, state)
+					}),
 				),
 			},
 			// Update schema to a new version
@@ -52,9 +56,11 @@ func TestAccSchemaDataSource_multipleVersions(t *testing.T) {
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrSet(datasourceName, "id"),
 					resource.TestCheckResourceAttr(datasourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(datasourceName, "schema", NormalizedJSON(updatedSchema)),
 					resource.TestCheckResourceAttr(datasourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(datasourceName, "compatibility_level", "BACKWARD"),
+					resource.TestCheckResourceAttrWith(datasourceName, "schema", func(state string) error {
+						return ValidateSchemaString(updatedSchema, state)
+					}),
 				),
 			},
 			// Validate updated version of the schema
@@ -63,10 +69,12 @@ func TestAccSchemaDataSource_multipleVersions(t *testing.T) {
 				Check: resource.ComposeAggregateTestCheckFunc(
 					resource.TestCheckResourceAttrSet(datasourceName, "id"),
 					resource.TestCheckResourceAttr(datasourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(datasourceName, "schema", NormalizedJSON(updatedSchema)),
 					resource.TestCheckResourceAttr(datasourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(datasourceName, "compatibility_level", "BACKWARD"),
 					resource.TestCheckResourceAttr(datasourceName, "version", "2"),
+					resource.TestCheckResourceAttrWith(datasourceName, "schema", func(state string) error {
+						return ValidateSchemaString(updatedSchema, state)
+					}),
 				),
 			},
 		},
@@ -102,7 +110,11 @@ resource "schemaregistry_schema" "test_01" {
     {
       "name": "f1",
       "type": "string"
-    }
+    },
+	{
+	  "name": "f2",
+	  "type": "string"
+	}
   ]
 })
 }
diff --git a/internal/provider/provider_test.go b/internal/provider/provider_test.go
index f368724..f41fecb 100644
--- a/internal/provider/provider_test.go
+++ b/internal/provider/provider_test.go
@@ -14,7 +14,7 @@ import (
 const (
 	testAccProviderVersion = "test"
 	testAccProviderType    = "schemaregistry"
-	redpandaContainerImage = "docker.redpanda.com/redpandadata/redpanda:v23.3.3"
+	redpandaContainerImage = "docker.redpanda.com/redpandadata/redpanda:v24.1.15"
 )
 
 var testAccProtoV6ProviderFactories = map[string]func() (tfprotov6.ProviderServer, error){
@@ -32,6 +32,7 @@ func TestMain(m *testing.M) {
 		redpanda.WithNewServiceAccount("superuser-1", "test"),
 		redpanda.WithSuperusers("superuser-1"),
 		redpanda.WithEnableSchemaRegistryHTTPBasicAuth(),
+		redpanda.WithBootstrapConfig("schema_registry_normalize_on_startup", true),
 	)
 	if err != nil {
 		log.Fatalf("failed to start container: %s", err)
diff --git a/internal/provider/resource_schema.go b/internal/provider/resource_schema.go
index 225ac02..8fe72af 100644
--- a/internal/provider/resource_schema.go
+++ b/internal/provider/resource_schema.go
@@ -193,24 +193,14 @@ func (r *schemaResource) Create(ctx context.Context, req resource.CreateRequest,
 		return
 	}
 
-	// Normalize the schema string
-	schemaString := plan.Schema.ValueString()
-	normalizedSchema, err := NormalizeJSON(schemaString, &resp.Diagnostics)
-	if err != nil {
-		resp.Diagnostics.AddError(
-			"Invalid JSON Schema",
-			fmt.Sprintf("Schema validation failed: %s", err),
-		)
-		return
-	}
-
 	// Generate API request body from plan
+	schemaString := plan.Schema.ValueString()
 	schemaType := ToSchemaType(plan.SchemaType.ValueString())
 	references := ToRegistryReferences(plan.Reference)
 	compatibilityLevel := ToCompatibilityLevelType(plan.CompatibilityLevel.ValueString())
 
 	// Create new schema resource
-	schema, err := r.client.CreateSchema(subject, normalizedSchema, schemaType, references...)
+	schema, err := r.client.CreateSchema(subject, schemaString, schemaType, references...)
 	if err != nil {
 		resp.Diagnostics.AddError(
 			"Error creating schema",
@@ -268,20 +258,10 @@ func (r *schemaResource) Read(ctx context.Context, req resource.ReadRequest, res
 	}
 
 	schemaType := FromSchemaType(schema.SchemaType())
-
-	// Normalize the schema string
 	schemaString := state.Schema.ValueString()
-	normalizedSchema, err := NormalizeJSON(schemaString, &resp.Diagnostics)
-	if err != nil {
-		resp.Diagnostics.AddError(
-			"Invalid JSON Schema",
-			fmt.Sprintf("Schema validation failed: %s", err),
-		)
-		return
-	}
 
 	// Update state with refreshed values
-	state.Schema = jsontypes.NewNormalizedValue(normalizedSchema)
+	state.Schema = jsontypes.NewNormalizedValue(schemaString)
 	state.SchemaID = types.Int64Value(int64(schema.ID()))
 	state.SchemaType = types.StringValue(schemaType)
 	state.Version = types.Int64Value(int64(schema.Version()))
@@ -306,25 +286,15 @@ func (r *schemaResource) Update(ctx context.Context, req resource.UpdateRequest,
 		return
 	}
 
-	// Normalize the schema string
-	schemaString := plan.Schema.ValueString()
-	normalizedSchema, err := NormalizeJSON(schemaString, &resp.Diagnostics)
-	if err != nil {
-		resp.Diagnostics.AddError(
-			"Invalid JSON Schema",
-			fmt.Sprintf("Schema validation failed: %s", err),
-		)
-		return
-	}
-
 	// Generate API request body from plan
+	schemaString := plan.Schema.ValueString()
 	subject := plan.Subject.ValueString()
 	references := ToRegistryReferences(plan.Reference)
 	schemaType := ToSchemaType(plan.SchemaType.ValueString())
 	compatibilityLevel := ToCompatibilityLevelType(plan.CompatibilityLevel.ValueString())
 
 	// Update existing schema
-	schema, err := r.client.CreateSchema(subject, normalizedSchema, schemaType, references...)
+	schema, err := r.client.CreateSchema(subject, schemaString, schemaType, references...)
 	if err != nil {
 		resp.Diagnostics.AddError(
 			"Error updating schema",
@@ -343,7 +313,7 @@ func (r *schemaResource) Update(ctx context.Context, req resource.UpdateRequest,
 	}
 
 	// Update state with refreshed values
-	plan.Schema = jsontypes.NewNormalizedValue(normalizedSchema)
+	plan.Schema = jsontypes.NewNormalizedValue(schemaString)
 	plan.SchemaID = types.Int64Value(int64(schema.ID()))
 	plan.Version = types.Int64Value(int64(schema.Version()))
 	plan.Reference = FromRegistryReferences(schema.References())
@@ -409,21 +379,11 @@ func (r *schemaResource) ImportState(ctx context.Context, req resource.ImportSta
 
 	schemaType := FromSchemaType(schema.SchemaType())
 
-	// Normalize the schema string
-	normalizedSchema, err := NormalizeJSON(schema.Schema(), &resp.Diagnostics)
-	if err != nil {
-		resp.Diagnostics.AddError(
-			"Invalid JSON Schema",
-			fmt.Sprintf("Schema validation failed: %s", err),
-		)
-		return
-	}
-
 	// Create state from retrieved schema
 	state := schemaResourceModel{
 		ID:                 types.StringValue(subject),
 		Subject:            types.StringValue(subject),
-		Schema:             jsontypes.NewNormalizedValue(normalizedSchema),
+		Schema:             jsontypes.NewNormalizedValue(schema.Schema()),
 		SchemaID:           types.Int64Value(int64(schema.ID())),
 		SchemaType:         types.StringValue(schemaType),
 		Version:            types.Int64Value(int64(schema.Version())),
diff --git a/internal/provider/resource_schema_test.go b/internal/provider/resource_schema_test.go
index b244d50..6313b2a 100644
--- a/internal/provider/resource_schema_test.go
+++ b/internal/provider/resource_schema_test.go
@@ -6,6 +6,7 @@ import (
 
 	"github.com/hashicorp/terraform-plugin-testing/helper/acctest"
 	"github.com/hashicorp/terraform-plugin-testing/helper/resource"
+	"github.com/hashicorp/terraform-plugin-testing/terraform"
 )
 
 const (
@@ -16,6 +17,10 @@ const (
         {
             "name": "f1",
             "type": "string"
+        },
+        {
+            "name": "f2",
+            "type": "string"
         }
     ]
 }`
@@ -34,6 +39,8 @@ const (
         }
     ]
 }`
+
+	expectedSchema = `{"type":"record","name":"TestUpdated","fields":[{"name":"f1","type":"string"},{"name":"f2","type":"int"}]}`
 )
 
 func TestAccSchemaResource_basic(t *testing.T) {
@@ -44,41 +51,57 @@ func TestAccSchemaResource_basic(t *testing.T) {
 	resource.ParallelTest(t, resource.TestCase{
 		ProtoV6ProviderFactories: testAccProtoV6ProviderFactories,
 		Steps: []resource.TestStep{
-			// Create, Read, ImportState testing
+			// Create and Read testing
 			{
 				Config: testAccSchemaResourceConfig_single(subjectName, ref01),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					// Verify schema attributes
 					resource.TestCheckResourceAttr(resourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(resourceName, "schema", NormalizedJSON(initialSchema)),
 					resource.TestCheckResourceAttr(resourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(resourceName, "compatibility_level", "NONE"),
 					resource.TestCheckResourceAttr(resourceName, "hard_delete", "false"),
 					resource.TestCheckResourceAttrSet(resourceName, "schema_id"),
 					resource.TestCheckResourceAttrSet(resourceName, "version"),
+					resource.TestCheckResourceAttrWith(resourceName, "schema", func(state string) error {
+						return ValidateSchemaString(initialSchema, state)
+					}),
 				),
 			},
-			// ImportState testing
-			{
-				ResourceName:            resourceName,
-				ImportState:             true,
-				ImportStateVerify:       true,
-				ImportStateVerifyIgnore: []string{},
-			},
-			// Update testing
+			// Update and Read testing
 			{
 				Config: testAccSchemaResourceConfig_singleUpdate(subjectName, ref01),
 				Check: resource.ComposeAggregateTestCheckFunc(
 					// Verify updated schema attributes
 					resource.TestCheckResourceAttr(resourceName, "subject", subjectName),
-					resource.TestCheckResourceAttr(resourceName, "schema", NormalizedJSON(updatedSchema)),
 					resource.TestCheckResourceAttr(resourceName, "schema_type", "AVRO"),
 					resource.TestCheckResourceAttr(resourceName, "compatibility_level", "BACKWARD"),
 					resource.TestCheckResourceAttr(resourceName, "hard_delete", "true"),
 					resource.TestCheckResourceAttrSet(resourceName, "schema_id"),
 					resource.TestCheckResourceAttrSet(resourceName, "version"),
+					resource.TestCheckResourceAttrWith(resourceName, "schema", func(state string) error {
+						return ValidateSchemaString(updatedSchema, state)
+					}),
 				),
 			},
+			// ImportState testing
+			{
+				ResourceName: resourceName,
+				ImportState:  true,
+				ImportStateCheck: func(states []*terraform.InstanceState) error {
+					if len(states) != 1 {
+						return fmt.Errorf("expected 1 state, got %d", len(states))
+					}
+					state := states[0]
+					if state.Attributes["subject"] != subjectName {
+						return fmt.Errorf("expected subject %s, got %s", subjectName, state.Attributes["subject"])
+					}
+					err := ValidateSchemaString(expectedSchema, state.Attributes["schema"])
+					if err != nil {
+						return fmt.Errorf("schema validation error: %v", err)
+					}
+					return nil
+				},
+			},
 		},
 	})
 }
@@ -101,12 +124,6 @@ func TestAccSchemaResource_withReferences(t *testing.T) {
 					resource.TestCheckResourceAttr(resourceName, "references.0.version", "1"),
 				),
 			},
-			{
-				ResourceName:            resourceName,
-				ImportState:             true,
-				ImportStateVerify:       true,
-				ImportStateVerifyIgnore: []string{},
-			},
 			{
 				Config: testAccSchemaResourceConfig_singleUpdate(subjectName, ref01),
 				Check: resource.ComposeAggregateTestCheckFunc(
@@ -168,7 +185,11 @@ resource "schemaregistry_schema" "test_01" {
     {
       "name": "f1",
       "type": "string"
-    }
+    },
+	{
+	  "name": "f2",
+	  "type": "string"
+	}
   ]
 })
 references = [
diff --git a/internal/provider/utils.go b/internal/provider/utils.go
index b403ac3..748b07e 100644
--- a/internal/provider/utils.go
+++ b/internal/provider/utils.go
@@ -1,12 +1,13 @@
 package provider
 
 import (
-	"encoding/json"
+	"context"
+	"errors"
 	"fmt"
 	"os"
 	"strings"
 
-	"github.com/hashicorp/terraform-plugin-framework/diag"
+	"github.com/hashicorp/terraform-plugin-framework-jsontypes/jsontypes"
 )
 
 // getEnvOrDefault returns the value of the configuration or the environment variable.
@@ -28,36 +29,26 @@ func ConfigCompose(config ...string) string {
 	return str.String()
 }
 
-// NormalizeJSON normalizes a JSON string by unmarshaling it into a Go data structure and then marshaling it back to a
-// JSON string. If diagnostics is provided, errors encountered during the process are added to it.
-func NormalizeJSON(jsonString string, diagnostics *diag.Diagnostics) (string, error) {
-	var jsonData interface{}
-	err := json.Unmarshal([]byte(jsonString), &jsonData)
-	if err != nil {
-		diagnostics.AddError(
-			"Invalid JSON",
-			fmt.Sprintf("Error unmarshaling JSON: %s", err),
-		)
-		return "", err
-	}
+func ValidateSchemaString(expected, actual string) error {
+	ctx := context.Background()
 
-	normalizedBytes, err := json.Marshal(jsonData)
-	if err != nil {
-		diagnostics.AddError(
-			"Normalization Error",
-			fmt.Sprintf("Error marshaling JSON: %s", err),
-		)
-		return "", err
-	}
+	expectedSchemaString := jsontypes.NewNormalizedValue(expected)
+	actualSchemaString := jsontypes.NewNormalizedValue(actual)
 
-	return string(normalizedBytes), nil
-}
+	equal, diags := expectedSchemaString.StringSemanticEquals(ctx, actualSchemaString)
 
-// NormalizedJSON is a JSON normalization helper function for tests.
-func NormalizedJSON(jsonString string) string {
-	normalized, err := NormalizeJSON(jsonString, nil)
-	if err != nil {
-		return ""
+	if !equal {
+		return fmt.Errorf("input schema does not match output. Expected: %s, Actual: %s", expected, actual)
 	}
-	return normalized
+
+	if diags.HasError() {
+		var sb strings.Builder
+		for _, d := range diags.Errors() {
+			sb.WriteString(d.Summary() + "\n")
+			sb.WriteString(d.Detail() + "\n")
+		}
+		return errors.New(sb.String())
+	}
+
+	return nil
 }