fix: initial e2e test + associated fixes

Author: Tomek

.github/workflows/go.yml

@@ -22,26 +22,10 @@ jobs:
         go-version: '1.20'
 
     - name: Build
-      run: go build -v ./...
+      run: make build
 
     - name: Test
-      run: go test -v ./...
-
-#    - id: gen_image_tag
-#      name: Replace slash to dash
-#      uses: mad9000/actions-find-and-replace-string@3
-#      with:
-#        source:  ${{ github.ref_name }}
-#        find: '/'
-#        replace: '-'
-#
-#    - name: Dockerize and push
-#      env:
-#        DOCKER_REGISTRY: TODO
-#        IMAGE_TAG: ${{ steps.gen_image_tag.outputs.value }}
-#      run: |
-#        docker build -t $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG .
-#        docker push $ECR_REGISTRY/$ECR_REPOSITORY:$IMAGE_TAG
+      run: make test
 
     - name: Log in to Docker Hub
       uses: docker/login-action@f4ef78c080cd8ba55a85445d5b36e214a81df20a

.gitignore

@@ -25,9 +25,10 @@ Dockerfile.cross
 *.swo
 *~
 
-# Local 
+# Local
 .vscode/
 .env
 examples/
 packages/
 vals.yaml
+/config/crd/bases

README.md

@@ -52,8 +52,6 @@ schemaRegistry:
 
 ## How to use
 
-Deploy __ConfigMap__
-
 ```yaml
 apiVersion: kafka-schema-operator.incubly/v2beta1
 kind: KafkaSchema

controllers/kafkaschema_controller.go

@@ -40,22 +40,22 @@ type KafkaSchemaReconciler struct {
 }
 
 func (r *KafkaSchemaReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
-	log := log.FromContext(ctx)
+	logger := log.FromContext(ctx)
 
 	schema := &kafkaschemaoperatorv2beta1.KafkaSchema{}
 	err := r.Get(ctx, req.NamespacedName, schema)
 	if err != nil {
 		if errors.IsNotFound(err) {
-			log.Info("Schema resource not found. Ignoring since object must be deleted")
+			logger.Info("Schema resource not found. Ignoring since object must be deleted")
 			return ctrl.Result{}, nil
 		}
-		log.Error(err, "Failed to get Schema resource")
+		logger.Error(err, "Failed to get Schema resource")
 		return ctrl.Result{}, err
 	}
 
-	srClient, err := schemareg.NewClient(&schema.SchemaRegistry, log)
+	srClient, err := schemareg.NewClient(&schema.SchemaRegistry, logger)
 	if err != nil {
-		log.Error(err, "Failed to instantiate Schema Registry Client")
+		logger.Error(err, "Failed to instantiate Schema Registry Client")
 		return ctrl.Result{}, err
 	}
 
@@ -75,23 +75,23 @@ func (r *KafkaSchemaReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 			controllerutil.RemoveFinalizer(schema, schemaFinilizers)
 			err = r.Update(ctx, schema)
 			if err != nil {
-				log.Error(err, "Failed to delete KafkaSchema from kubernetes: "+schema.Name)
+				logger.Error(err, "Failed to delete KafkaSchema from kubernetes: "+schema.Name)
 				return ctrl.Result{}, err
 			}
-			log.Info("KafkaSchema CR was deleted: " + schema.Name)
+			logger.Info("KafkaSchema CR was deleted: " + schema.Name)
 
 			err := srClient.DeleteSubject(schemaKey)
 			if err != nil {
-				log.Error(err, "Failed to delete subject for key schema from registry: "+schemaKey)
+				logger.Error(err, "Failed to delete subject for key schema from registry: "+schemaKey)
 				return ctrl.Result{}, err
 			}
 
 			err = srClient.DeleteSubject(schemaValue)
 			if err != nil {
-				log.Error(err, "Failed to delete subject for value schema from registry: "+schemaValue)
+				logger.Error(err, "Failed to delete subject for value schema from registry: "+schemaValue)
 				return ctrl.Result{}, err
 			}
-			log.Info("Schema was removed from registry")
+			logger.Info("Schema was removed from registry")
 			return ctrl.Result{}, nil
 		}
 		return ctrl.Result{}, nil
@@ -101,60 +101,59 @@ func (r *KafkaSchemaReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		controllerutil.AddFinalizer(schema, schemaFinilizers)
 		err = r.Update(ctx, schema)
 		if err != nil {
-			log.Info("Failed to update finalizers for CR: " + schema.Name)
+			logger.Info("Failed to update finalizers for CR: " + schema.Name)
 			return ctrl.Result{}, err
 		}
-		log.Info("Finalizers are set for CR: " + schema.Name)
+		logger.Info("Finalizers are set for CR: " + schema.Name)
 	}
 
 	err = srClient.RegisterSchema(
 		schemaKey,
 		schemareg.RegisterSchemaReq{
-			Schema: `{\"type\": \"` + schema.Spec.SchemaSerializer + `\"}`,
+			Schema: `{"type": "` + schema.Spec.SchemaSerializer + `"}`,
 		})
 	if err != nil {
-		log.Error(err, "Failed to update schema registry")
+		logger.Error(err, "Failed to update schema registry")
 		return reconcileResult, err
 	}
-	log.Info("Schema key was published: " + schemaKey)
-
-	cfgData := schema.Spec.Data.Schema
-	cfgData = strings.ReplaceAll(cfgData, "\n", "")
-	cfgData = strings.ReplaceAll(cfgData, "\t", "")
-	cfgData = strings.ReplaceAll(cfgData, " ", "")
-	cfgData = strings.ReplaceAll(cfgData, `"`, `\"`)
-	cfgData = strings.Replace(cfgData, `\"{`, `"{`, 1)
-	cfgData = strings.Replace(cfgData, `}\"`, `}"`, -1)
+	logger.Info("Schema key was published: " + schemaKey)
 
 	err = srClient.RegisterSchema(
 		schemaValue,
 		schemareg.RegisterSchemaReq{
-			Schema:     cfgData,
+			Schema:     schema.Spec.Data.Schema,
 			SchemaType: strings.ToUpper(schema.Spec.Data.Format),
 		})
 	if err != nil {
-		log.Error(err, "Failed to update schema registry")
+		logger.Error(err, "Failed to update schema registry")
 		return reconcileResult, err
 	}
 
-	var schemaCompatibilityPayload strings.Builder
-	schemaCompatibilityPayload.WriteString(`{"compatibility": "`)
-	schemaCompatibilityPayload.WriteString(schema.Spec.Data.Compatibility)
-	schemaCompatibilityPayload.WriteString(`"}`)
+	err = srClient.SetCompatibilityMode(
+		schemaValue,
+		schemareg.SetCompatibilityModeReq{
+			Compatibility: schema.Spec.Data.Compatibility,
+		},
+	)
 
-	err = srClient.SetCompatibilityMode(schemaValue, schemaCompatibilityPayload.String())
 	if err != nil {
-		log.Error(err, "Failed to update schema compatibility for value")
+		logger.Error(err, "Failed to update schema compatibility for value")
 		return reconcileResult, err
 	}
 
-	err = srClient.SetCompatibilityMode(schemaKey, schemaCompatibilityPayload.String())
+	err = srClient.SetCompatibilityMode(
+		schemaKey,
+		schemareg.SetCompatibilityModeReq{
+			Compatibility: schema.Spec.Data.Compatibility,
+		},
+	)
+
 	if err != nil {
-		log.Error(err, "Failed to update schema compatibility for key")
+		logger.Error(err, "Failed to update schema compatibility for key")
 		return reconcileResult, err
 	}
 
-	log.Info("Schema value was published: " + schemaValue)
+	logger.Info("Schema value was published: " + schemaValue)
 	return reconcileResult, nil
 }
 

controllers/kafkaschema_controller_test.go

@@ -0,0 +1,88 @@
+package controllers
+
+import (
+	"context"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"kafka-schema-operator/api/v2beta1"
+	schemareg_mock "kafka-schema-operator/schemareg-mock"
+	"strconv"
+	"time"
+)
+
+const ResourceNs = "default"
+
+var _ = Describe("KafkaschemaController", func() {
+
+	BeforeEach(func() {
+		srMock.Clear()
+	})
+	Context("When creating Schema", func() {
+
+		It("Should create schema", func() {
+			topicName := "testenv.testservice.testevent." + strconv.FormatInt(time.Now().UnixMilli(), 10)
+			schemaRes := v2beta1.KafkaSchema{
+				TypeMeta: metav1.TypeMeta{
+					Kind:       "KafkaSchema",
+					APIVersion: "kafka-schema-operator.incubly/v2beta1",
+				},
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      topicName,
+					Namespace: ResourceNs,
+				},
+				Spec: v2beta1.KafkaSchemaSpec{
+					Name:                  topicName,
+					SchemaSerializer:      "string",
+					AutoReconciliation:    false,
+					TerminationProtection: false,
+					Data: v2beta1.KafkaSchemaData{
+						Schema: `{
+						  "namespace": "testing",
+						  "type": "record",
+						  "name": "testing",
+						  "fields": [
+						    {"name": "id", "type": "string"},
+						    {"name": "email", "type": "string"}
+						  ]
+						}`,
+						Format:        "AVRO",
+						Compatibility: "BACKWARD",
+					},
+				},
+			}
+
+			By("When: creating KafkaSchema resource")
+			ctx := context.Background()
+			Expect(k8sClient.Create(ctx, &schemaRes)).Should(Succeed())
+
+			By("Then: resource is created in K8S")
+			schemaLookupKey := types.NamespacedName{Name: topicName, Namespace: ResourceNs}
+			createdResource := &v2beta1.KafkaSchema{}
+			Eventually(func() error {
+				return k8sClient.Get(ctx, schemaLookupKey, createdResource)
+			}).Should(Succeed())
+
+			Expect(createdResource.Spec.Name).Should(Equal(topicName))
+			// TODO: verify resource
+
+			By("And: subjects and schemas for topic key & value are registered in schema registry")
+
+			Eventually(func() map[string]*schemareg_mock.Subject {
+				return srMock.Subjects
+			},
+				time.Second*10,
+				time.Millisecond*100,
+			).Should(
+				And(
+					HaveKey(topicName+"-key"),
+					HaveKey(topicName+"-value"),
+				),
+			)
+			Expect(srMock.Subjects[topicName+"-value"].SchemaRefs).Should(HaveLen(1))
+			// TODO: verify schema
+		})
+
+	})
+})

controllers/suite_test.go

@@ -17,11 +17,18 @@ limitations under the License.
 package controllers
 
 import (
-	"path/filepath"
-	"testing"
-
+	"errors"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/gexec"
+	"github.com/onsi/gomega/ghttp"
+	schemaregmock "kafka-schema-operator/schemareg-mock"
+	"net/url"
+	"os"
+	"path/filepath"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"testing"
+	"time"
 
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
@@ -37,9 +44,13 @@ import (
 // These tests use Ginkgo (BDD-style Go testing framework). Refer to
 // http://onsi.github.io/ginkgo/ to learn more about Ginkgo.
 
-var cfg *rest.Config
-var k8sClient client.Client
-var testEnv *envtest.Environment
+var (
+	cfg          *rest.Config
+	k8sClient    client.Client
+	testEnv      *envtest.Environment
+	srMockServer *ghttp.Server
+	srMock       *schemaregmock.SchemaRegMock
+)
 
 func TestAPIs(t *testing.T) {
 	RegisterFailHandler(Fail)
@@ -51,6 +62,8 @@ var _ = BeforeSuite(func() {
 	logf.SetLogger(zap.New(zap.WriteTo(GinkgoWriter), zap.UseDevMode(true)))
 
 	By("bootstrapping test environment")
+	Expect(setupKubeBuilderAssets()).To(Succeed())
+
 	testEnv = &envtest.Environment{
 		CRDDirectoryPaths:     []string{filepath.Join("..", "config", "crd", "bases")},
 		ErrorIfCRDPathMissing: true,
@@ -71,10 +84,63 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
 
+	By("Starting Schema Registry Mock")
+	srMock = schemaregmock.NewSchemaRegMock(
+		map[string]schemaregmock.SchemaCompatibilityPolicy{},
+		ctrl.Log,
+	)
+	srMockServer = srMock.GetServer()
+	srMockServerUrl, err := url.ParseRequestURI(srMockServer.HTTPTestServer.URL)
+	Expect(err).ToNot(HaveOccurred())
+	Expect(os.Setenv("SCHEMA_REGISTRY_HOST", srMockServerUrl.Hostname())).To(Succeed())
+	Expect(os.Setenv("SCHEMA_REGISTRY_PORT", srMockServerUrl.Port())).To(Succeed())
+
+	By("Run Controllers")
+	options := ctrl.Options{
+		Scheme:    scheme.Scheme,
+		Port:      9443,
+		Namespace: "default",
+	}
+
+	k8sManager, err := ctrl.NewManager(cfg, options)
+	Expect(err).ToNot(HaveOccurred())
+
+	err = (&KafkaSchemaReconciler{
+		Client: k8sManager.GetClient(),
+		Scheme: k8sManager.GetScheme(),
+	}).SetupWithManager(k8sManager)
+	Expect(err).ToNot(HaveOccurred())
+
+	go func() {
+		defer GinkgoRecover()
+		err = k8sManager.Start(ctrl.SetupSignalHandler())
+		Expect(err).ToNot(HaveOccurred(), "failed to run manager")
+		gexec.KillAndWait(4 * time.Second)
+
+		// Teardown the test environment once controller is finished.
+		// Otherwise, from Kubernetes 1.21+, teardown timeouts waiting on
+		// kube-apiserver to return
+		By("tearing down the test environment")
+		err := testEnv.Stop()
+		Expect(err).ToNot(HaveOccurred())
+		srMockServer.Close()
+	}()
 })
 
-var _ = AfterSuite(func() {
-	By("tearing down the test environment")
-	err := testEnv.Stop()
+func setupKubeBuilderAssets() error {
+	kubeBuilderAssets, err := os.ReadDir("../bin/k8s")
 	Expect(err).NotTo(HaveOccurred())
-})
+	var kubeBuilderAssetsPath string
+	for _, dirEntry := range kubeBuilderAssets {
+		if dirEntry.IsDir() {
+			kubeBuilderAssetsPath = "../bin/k8s/" + dirEntry.Name()
+			break
+		}
+	}
+	if len(kubeBuilderAssetsPath) == 0 {
+		return errors.New("kubeBuilderAssetsPath not found")
+	}
+
+	ctrl.Log.Info("Setting env KUBEBUILDER_ASSETS=" + kubeBuilderAssetsPath)
+	return os.Setenv("KUBEBUILDER_ASSETS", kubeBuilderAssetsPath)
+}