increase testing coverage

pkg/pubsub/adapter/converters/pubsub.go

@@ -57,7 +57,7 @@ func convertPubSub(ctx context.Context, msg *cepubsub.Message, sendMode ModeType
 		logger := logging.FromContext(ctx).With(zap.Any("event.id", event.ID()))
 		// set the content type to something that can be handled by codec.go
 		event.SetDataContentType(cloudevents.ApplicationJSON)
-		msg := &PubSubMessage{
+		msg := &pubSubMessage{
 			ID:          event.ID(),
 			Attributes:  msg.Attributes,
 			PublishTime: event.Time(),
@@ -73,7 +73,7 @@ func convertPubSub(ctx context.Context, msg *cepubsub.Message, sendMode ModeType
 			msg.Data = raw
 		}
 
-		if err := event.SetData(&PushMessage{
+		if err := event.SetData(&pushMessage{
 			Subscription: tx.Subscription,
 			Message:      msg,
 		}); err != nil {
@@ -94,16 +94,16 @@ func convertPubSub(ctx context.Context, msg *cepubsub.Message, sendMode ModeType
 	return &event, nil
 }
 
-// PushMessage represents the format Pub/Sub uses to push events.
-type PushMessage struct {
+// pushMessage represents the format Pub/Sub uses to push events.
+type pushMessage struct {
 	// Subscription is the subscription ID that received this Message.
 	Subscription string `json:"subscription"`
 	// Message holds the Pub/Sub message contents.
-	Message *PubSubMessage `json:"message,omitempty"`
+	Message *pubSubMessage `json:"message,omitempty"`
 }
 
 // PubSubMessage matches the inner message format used by Push Subscriptions.
-type PubSubMessage struct {
+type pubSubMessage struct {
 	// ID identifies this message. This ID is assigned by the server and is
 	// populated for Messages obtained from a subscription.
 	// This field is read-only.

pkg/pubsub/adapter/converters/pubsub_test.go

@@ -52,7 +52,7 @@ func TestConvertCloudPubSub(t *testing.T) {
 			return pubSubCloudEvent(map[string]string{
 				"attribute1": "value1",
 				"attribute2": "value2",
-			})
+			}, "")
 		},
 	}, {
 		name: "upper case attributes",
@@ -68,7 +68,7 @@ func TestConvertCloudPubSub(t *testing.T) {
 			return pubSubCloudEvent(map[string]string{
 				"attribute1": "value1",
 				"attribute2": "value2",
-			})
+			}, "")
 		},
 	}, {
 		name: "only setting valid alphanumeric attribute",
@@ -83,7 +83,24 @@ func TestConvertCloudPubSub(t *testing.T) {
 		wantEventFn: func() *cloudevents.Event {
 			return pubSubCloudEvent(map[string]string{
 				"attribute1": "value1",
-			})
+			}, "")
+		},
+	}, {
+		name: "schema as attribute",
+		message: &cepubsub.Message{
+			Data: []byte("test data"),
+			Attributes: map[string]string{
+				"attribute1": "value1",
+				"attribute2": "value2",
+				"schema":     "schema_val",
+			},
+		},
+		sendMode: Binary,
+		wantEventFn: func() *cloudevents.Event {
+			return pubSubCloudEvent(map[string]string{
+				"attribute1": "value1",
+				"attribute2": "value2",
+			}, "schema_val")
 		},
 	}, {
 		name: "Push mode with non valid alphanumeric attribute",
@@ -139,7 +156,7 @@ func TestConvertCloudPubSub(t *testing.T) {
 	}
 }
 
-func pubSubCloudEvent(extensions map[string]string) *cloudevents.Event {
+func pubSubCloudEvent(extensions map[string]string, schema string) *cloudevents.Event {
 	e := cloudevents.NewEvent(cloudevents.VersionV1)
 	e.SetID("id")
 	e.SetSource(v1alpha1.CloudPubSubSourceEventSource("testproject", "testtopic"))
@@ -151,6 +168,9 @@ func pubSubCloudEvent(extensions map[string]string) *cloudevents.Event {
 	for k, v := range extensions {
 		e.SetExtension(k, v)
 	}
+	if schema != "" {
+		e.SetDataSchema(schema)
+	}
 	return &e
 }