chipingress: add RegisterSchemas method and noop client (#1646)

* chipingress: add RegisterSchemas method; add noop client implementation

* export chipingress noop client

Author: pkcll

pkg/chipingress/client.go

@@ -17,6 +17,7 @@ import (
 	"google.golang.org/grpc/metadata"
 
 	ceformat "github.com/cloudevents/sdk-go/binding/format/protobuf/v2"
+	cepb "github.com/cloudevents/sdk-go/binding/format/protobuf/v2/pb"
 	ce "github.com/cloudevents/sdk-go/v2"
 
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
@@ -27,6 +28,7 @@ import (
 type Client interface {
 	pb.ChipIngressClient
 	Close() error
+	RegisterSchemas(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error)
 }
 
 type client struct {
@@ -128,6 +130,23 @@ func (c *client) Close() error {
 	return c.conn.Close()
 }
 
+// RegisterSchemas registers one or more schemas with the Chip Ingress service.
+func (c *client) RegisterSchemas(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error) {
+	request := &pb.RegisterSchemaRequest{Schemas: schemas}
+
+	resp, err := c.client.RegisterSchema(ctx, request)
+	if err != nil {
+		return nil, fmt.Errorf("failed to register schema: %w", err)
+	}
+
+	registeredMap := make(map[string]int)
+	for _, schema := range resp.Registered {
+		registeredMap[schema.Subject] = int(schema.Version)
+	}
+
+	return registeredMap, nil
+}
+
 // WithBasicAuth sets the basic-auth credentials for the ChipIngress service.
 // Default is to require TLS for security.
 func WithBasicAuth(user, pass string) Opt {
@@ -281,3 +300,44 @@ func EventsToBatch(events []CloudEvent) (*CloudEventBatch, error) {
 	}
 	return batch, nil
 }
+
+var _ Client = (*NoopClient)(nil)
+
+// NoopClient is a no-op implementation of the Client interface.
+// All methods return successfully without performing any actual operations.
+type NoopClient struct{}
+
+// Close is a no-op
+func (NoopClient) Close() error {
+	return nil
+}
+
+// Ping is a no-op
+func (NoopClient) Ping(ctx context.Context, in *pb.EmptyRequest, opts ...grpc.CallOption) (*pb.PingResponse, error) {
+	return &pb.PingResponse{Message: "pong"}, nil
+}
+
+// Publish is a no-op
+func (NoopClient) Publish(ctx context.Context, in *cepb.CloudEvent, opts ...grpc.CallOption) (*pb.PublishResponse, error) {
+	return &pb.PublishResponse{}, nil
+}
+
+// PublishBatch is a no-op
+func (NoopClient) PublishBatch(ctx context.Context, in *pb.CloudEventBatch, opts ...grpc.CallOption) (*pb.PublishResponse, error) {
+	return &pb.PublishResponse{}, nil
+}
+
+// StreamEvents is a no-op
+func (NoopClient) StreamEvents(ctx context.Context, opts ...grpc.CallOption) (grpc.BidiStreamingClient[pb.StreamEventsRequest, pb.StreamEventsResponse], error) {
+	return nil, nil
+}
+
+// RegisterSchema is a no-op
+func (NoopClient) RegisterSchema(ctx context.Context, in *pb.RegisterSchemaRequest, opts ...grpc.CallOption) (*pb.RegisterSchemaResponse, error) {
+	return &pb.RegisterSchemaResponse{}, nil
+}
+
+// RegisterSchemas is a no-op
+func (NoopClient) RegisterSchemas(ctx context.Context, schemas ...*pb.Schema) (map[string]int, error) {
+	return make(map[string]int), nil
+}

pkg/chipingress/client_test.go

@@ -14,6 +14,7 @@ import (
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/protobuf/proto"
 
+	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/mocks"
 	"github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb"
 )
 
@@ -56,6 +57,33 @@ func TestClient(t *testing.T) {
 		assert.NotNil(t, client)
 	})
 
+	t.Run("NoopClient", func(t *testing.T) {
+		client := &NoopClient{}
+		assert.NotNil(t, client)
+
+		// Test that it implements the Client interface
+		var _ Client = client
+
+		// Test Close returns no error
+		err := client.Close()
+		assert.NoError(t, err)
+
+		// Test Ping returns success
+		pingResp, err := client.Ping(context.Background(), &pb.EmptyRequest{})
+		assert.NoError(t, err)
+		assert.NotNil(t, pingResp)
+		assert.Equal(t, "pong", pingResp.Message)
+
+		// Test RegisterSchemas returns empty map
+		schemas := []*pb.Schema{
+			{Subject: "test", Schema: `{"test":"value"}`, Format: 1},
+		}
+		result, err := client.RegisterSchemas(context.Background(), schemas...)
+		assert.NoError(t, err)
+		assert.NotNil(t, result)
+		assert.Empty(t, result)
+	})
+
 }
 
 func TestNewEvent(t *testing.T) {
@@ -619,3 +647,62 @@ func TestNewClientWithTLS(t *testing.T) {
 		assert.NotNil(t, client)
 	}
 }
+
+func TestClient_RegisterSchemas(t *testing.T) {
+	t.Run("successfully registers schemas", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		mockClient.EXPECT().RegisterSchema(
+			context.Background(),
+			&pb.RegisterSchemaRequest{
+				Schemas: []*pb.Schema{
+					{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+					{Subject: "schema2", Schema: `{"type":"record","name":"Test2","fields":[{"name":"field2"}]}`, Format: 2},
+				},
+			},
+		).Return(&pb.RegisterSchemaResponse{
+			Registered: []*pb.RegisteredSchema{
+				{Subject: "schema1", Version: 1},
+				{Subject: "schema2", Version: 2},
+			},
+		}, nil)
+
+		client := &client{
+			client: mockClient,
+			conn:   nil,
+		}
+
+		schemas := []*pb.Schema{
+			{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+			{Subject: "schema2", Schema: `{"type":"record","name":"Test2","fields":[{"name":"field2"}]}`, Format: 2},
+		}
+
+		result, err := client.RegisterSchemas(context.Background(), schemas...)
+		assert.NoError(t, err)
+		assert.Equal(t, map[string]int{"schema1": 1, "schema2": 2}, result)
+	})
+
+	t.Run("returns error when registration fails", func(t *testing.T) {
+		mockClient := mocks.NewClient(t)
+		mockClient.EXPECT().RegisterSchema(
+			context.Background(),
+			&pb.RegisterSchemaRequest{
+				Schemas: []*pb.Schema{
+					{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+				},
+			},
+		).Return(nil, fmt.Errorf("registration failed"))
+
+		client := &client{
+			client: mockClient,
+			conn:   nil,
+		}
+
+		schemas := []*pb.Schema{
+			{Subject: "schema1", Schema: `{"type":"record","name":"Test","fields":[{"name":"field1"}]}`, Format: 1},
+		}
+
+		result, err := client.RegisterSchemas(context.Background(), schemas...)
+		assert.Nil(t, result)
+		assert.EqualError(t, err, "failed to register schema: registration failed")
+	})
+}