Repository: hamba/avro
Pull request #229 (2023-01-10 18:17:30)

Registry compatibility level
Wrote code for compatibility level putting and getting. For the testing a bit of guidance would be well accepted! I usually set my tests with a container running, but I couldn't manage to run the whole zookeper+kafka_broker+confluent_schema_registry using [testcontainers-go](https://github.com/testcontainers/testcontainers-go)

Thanks for the attention,
Nuc

nrwiersma: Hi.

For tests, the standard is unit tests. An example can be found here: https://github.com/hamba/avro/blob/main/registry/client_test.go#L110.

You code need to pass the linter as well.
nucccc: In the latest commit i made my best efforts to come towards the requests of the linter, and added some unit tests.

To me it appeared the linter required comments to end with a period, I didn't do that, obtaining an error. I also made my best to avoid warnings by setting the beginning of the comments as the name of the method/function.

I also added a unite test for each client method proposed, and a unit test for the function verifying if a compatibility level is valid according to confluent docs.

I hope the work makes sense, and I would kindly ask you if it is possible torerun the checks.

Thank you in advance,
Nuc
nucccc: I think I solved almost everything required.

I leave a point open:
- the functions setting the compatibility level actually return the string, complying with what the API does (shrug!). In my personal opinion you can decide what you prefer, it's up to you.

Nuc
nucccc: Made my best to come towards all actual revisions

Patch:
diff --git a/registry/client.go b/registry/client.go
index 2bb9037..2a6e710 100644
--- a/registry/client.go
+++ b/registry/client.go
@@ -352,3 +352,98 @@ func (e Error) Error() string {
 
 	return "registry error: " + strconv.Itoa(e.StatusCode)
 }
+
+// Compatibility levels.
+const (
+	BackwardCL           string = "BACKWARD"
+	BackwardTransitiveCL string = "BACKWARD_TRANSITIVE"
+	ForwardCL            string = "FORWARD"
+	ForwardTransitiveCL  string = "FORWARD_TRANSITIVE"
+	FullCL               string = "FULL"
+	FullTransitiveCL     string = "FULL_TRANSITIVE"
+	NoneCL               string = "NONE"
+)
+
+// ValidateCompatibilityLevel returns whether or not a compatibility level
+// is valid according to the ones described in:
+// https://docs.confluent.io/platform/current/schema-registry/develop/api.html#compatibility
+// .
+
+func validateCompatibilityLevel(lvl string) error {
+	switch lvl {
+	case BackwardCL, BackwardTransitiveCL, ForwardCL, ForwardTransitiveCL, FullCL, FullTransitiveCL, NoneCL:
+		return nil
+	default:
+		return fmt.Errorf("invalid compatibility level %s", lvl)
+	}
+}
+
+// compatibilityPayload is the response body obtained from the schema registry when updating a compatibility level.
+type compatibilityPayload struct {
+	Compatibility string `json:"compatibility"`
+}
+
+// SetGlobalCompatibilityLevel sets the global compatibility level of the registry.
+func (c *Client) SetGlobalCompatibilityLevel(
+	ctx context.Context,
+	lvl string,
+) error {
+	err := validateCompatibilityLevel(lvl)
+	if err != nil {
+		return err
+	}
+	var payload compatibilityPayload
+	inPayload := compatibilityPayload{Compatibility: lvl}
+	err = c.request(ctx, http.MethodPut, "config", inPayload, &payload)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// SetCompatibilityLevel sets the compatibility level of a subject.
+func (c *Client) SetCompatibilityLevel(
+	ctx context.Context,
+	subject, lvl string,
+) error {
+	err := validateCompatibilityLevel(lvl)
+	if err != nil {
+		return err
+	}
+	var payload compatibilityPayload
+	inPayload := compatibilityPayload{Compatibility: lvl}
+	err = c.request(ctx, http.MethodPut, "config/"+subject, inPayload, &payload)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// GetGlobalCompatibilityLevel gets the global compatibility level.
+func (c *Client) GetGlobalCompatibilityLevel(
+	ctx context.Context,
+) (string, error) {
+	var payload compatibilityPayload
+	err := c.request(ctx, http.MethodGet, "config", nil, &payload)
+	if err != nil {
+		return "", err
+	}
+
+	return payload.Compatibility, nil
+}
+
+// GetCompatibilityLevel gets the global compatibility level of a subject.
+func (c *Client) GetCompatibilityLevel(
+	ctx context.Context,
+	subject string,
+) (string, error) {
+	var payload compatibilityPayload
+	err := c.request(ctx, http.MethodGet, "config/"+subject, nil, &payload)
+	if err != nil {
+		return "", err
+	}
+
+	return payload.Compatibility, nil
+}


Test patch:
diff --git a/registry/client_internal_test.go b/registry/client_internal_test.go
index 2aa146b..54980cf 100644
--- a/registry/client_internal_test.go
+++ b/registry/client_internal_test.go
@@ -5,6 +5,7 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
 )
 
 func TestNewClient_WithHTTPClient(t *testing.T) {
@@ -22,3 +23,19 @@ func TestNewClient_WithBasicAuth(t *testing.T) {
 
 	assert.Equal(t, client.creds, creds)
 }
+
+func TestFunc_validateCompatibilityLevel(t *testing.T) {
+	cls := []string{
+		BackwardCL,
+		BackwardTransitiveCL,
+		ForwardCL,
+		ForwardTransitiveCL,
+		FullCL,
+		FullTransitiveCL,
+		NoneCL,
+	}
+	for _, cl := range cls {
+		require.NoError(t, validateCompatibilityLevel(cl))
+	}
+	assert.Equal(t, "invalid compatibility level BOH", validateCompatibilityLevel("BOH").Error())
+}
diff --git a/registry/client_test.go b/registry/client_test.go
index 7d72726..b0bf1bf 100644
--- a/registry/client_test.go
+++ b/registry/client_test.go
@@ -3,6 +3,7 @@ package registry_test
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"io"
 	"net/http"
 	"net/http/httptest"
@@ -599,3 +600,154 @@ func TestError_ErrorEmptyMEssage(t *testing.T) {
 
 	assert.Equal(t, "registry error: 404", str)
 }
+
+func TestClient_GetGlobalCompatibilityLevel(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "GET", r.Method)
+		assert.Equal(t, "/config", r.URL.Path)
+
+		_, _ = w.Write([]byte(`{
+			"compatibility": "FULL"
+		  }`))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	compatibilityLevel, err := client.GetGlobalCompatibilityLevel(context.Background())
+
+	require.NoError(t, err)
+	assert.Equal(t, registry.FullCL, compatibilityLevel)
+}
+
+func TestClient_GetGlobalCompatibilityLevelError(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(500)
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	_, err := client.GetGlobalCompatibilityLevel(context.Background())
+
+	assert.Error(t, err)
+}
+
+func TestClient_GetCompatibilityLevel(t *testing.T) {
+	subject := "boh_subj"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "GET", r.Method)
+		assert.Equal(t, "/config/"+subject, r.URL.Path)
+
+		_, _ = w.Write([]byte(`{
+			"compatibility": "FULL"
+		  }`))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	compatibilityLevel, err := client.GetCompatibilityLevel(context.Background(), subject)
+
+	require.NoError(t, err)
+	assert.Equal(t, registry.FullCL, compatibilityLevel)
+}
+
+func TestClient_GetCompatibilityLevelError(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(500)
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	_, err := client.GetCompatibilityLevel(context.Background(), "boh")
+
+	assert.Error(t, err)
+}
+
+func TestClient_SetGlobalCompatibilityLevel(t *testing.T) {
+	cl := registry.BackwardCL
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "PUT", r.Method)
+		assert.Equal(t, "/config", r.URL.Path)
+
+		_, _ = w.Write([]byte((fmt.Sprintf("{\"compatibility\":\"%s\"}", cl))))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetGlobalCompatibilityLevel(context.Background(), cl)
+
+	require.NoError(t, err)
+}
+
+func TestClient_SetGlobalCompatibilityLevelError(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(500)
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetGlobalCompatibilityLevel(context.Background(), registry.BackwardCL)
+
+	assert.Error(t, err)
+}
+
+func TestClient_SetGlobalCompatibilityLevelHandlesInvalidLevel(t *testing.T) {
+	invalid_cl := "INVALID_STUFF"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "PUT", r.Method)
+		assert.Equal(t, "/config", r.URL.Path)
+
+		_, _ = w.Write([]byte((fmt.Sprintf("{\"compatibility\":\"%s\"}", invalid_cl))))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetGlobalCompatibilityLevel(context.Background(), invalid_cl)
+
+	assert.Error(t, err)
+}
+
+func TestClient_SetCompatibilityLevel(t *testing.T) {
+	cl := registry.BackwardCL
+	subject := "boh_subj"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "PUT", r.Method)
+		assert.Equal(t, "/config/"+subject, r.URL.Path)
+
+		_, _ = w.Write([]byte((fmt.Sprintf("{\"compatibility\":\"%s\"}", cl))))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetCompatibilityLevel(context.Background(), subject, cl)
+
+	require.NoError(t, err)
+}
+
+func TestClient_SetCompatibilityLevelError(t *testing.T) {
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(500)
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetCompatibilityLevel(context.Background(), "boh", registry.BackwardCL)
+
+	assert.Error(t, err)
+}
+
+func TestClient_SetCompatibilityLevelHandlesInvalidLevel(t *testing.T) {
+	invalid_cl := "INVALID_STUFF"
+	subject := "boh"
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "PUT", r.Method)
+		assert.Equal(t, "/config/"+subject, r.URL.Path)
+
+		_, _ = w.Write([]byte((fmt.Sprintf("{\"compatibility\":\"%s\"}", invalid_cl))))
+	}))
+	defer s.Close()
+	client, _ := registry.NewClient(s.URL)
+
+	err := client.SetCompatibilityLevel(context.Background(), subject, invalid_cl)
+
+	assert.Error(t, err)
+}