add test for rejecting malformed CE
@@ -73,8 +73,7 @@ func DataPlaneIngress(brokerName string) *feature.Feature {
Must("The Broker MUST respond with a 200-level HTTP status code if a produce request is accepted.",
todo).
Must("If a Broker receives a produce request and is unable to parse a valid CloudEvent, then it MUST reject the request with HTTP status code `400 Bad Request`.",
- todo)
-
+ brokerRejectsMalformedCE)
return f
}
@@ -219,7 +218,7 @@ func brokerRejectsUnknownCEVersion(ctx context.Context, t feature.T) {
eventshub.InputHeader("ce-type", "sometype"),
eventshub.InputHeader("ce-source", "400.request.sender.test.knative.dev"),
eventshub.InputHeader("ce-id", uuid),
- eventshub.InputBody("{}}"),
+ eventshub.InputBody("{}"),
)(ctx, t)
store := eventshub.StoreFromContext(ctx, source)
@@ -259,7 +258,7 @@ func brokerRejectsGetRequest(ctx context.Context, t feature.T) {
eventshub.InputHeader("ce-type", "sometype"),
eventshub.InputHeader("ce-source", "400.request.sender.test.knative.dev"),
eventshub.InputHeader("ce-id", uuid),
- eventshub.InputBody("{}}"),
+ eventshub.InputBody("{}"),
eventshub.InputMethod("GET"),
)(ctx, t)
@@ -279,6 +278,46 @@ func brokerRejectsGetRequest(ctx context.Context, t feature.T) {
}
}
+func brokerRejectsMalformedCE(ctx context.Context, t feature.T) {
+ brokerName := state.GetStringOrFail(ctx, t, "brokerName")
+
+ u, err := broker.Address(ctx, brokerName)
+ if err != nil || u == nil {
+ t.Error("failed to get the address of the broker", brokerName, err)
+ }
+
+ uuids := []string{
+ uuid.New().String(),
+ }
+ for _, uuid := range uuids {
+ // We need to use a different source name, otherwise, it will try to update
+ // the pod, which is immutable.
+ source := feature.MakeRandomK8sName("source")
+ eventshub.Install(source,
+ eventshub.StartSenderToResource(broker.Gvr(), brokerName),
+ // Missing version here, should be rejected.
+ eventshub.InputHeader("ce-type", "sometype"),
+ eventshub.InputHeader("ce-source", "400.request.sender.test.knative.dev"),
+ eventshub.InputHeader("ce-id", uuid),
+ eventshub.InputBody("{}"),
+ )(ctx, t)
+
+ store := eventshub.StoreFromContext(ctx, source)
+ // We are looking for two events, one of them is the sent event and the other
+ // is Response, so correlate them first. We want to make sure the event was sent and that the
+ // response was what was expected.
+ // Note: We pass in "" for the match ID because when we construct the headers manually
+ // above, they do not get stuff into the sent/response SentId fields.
+ events := correlate(store.AssertAtLeast(2, sentEventMatcher("")))
+ for _, e := range events {
+ // Make sure HTTP response code is 400
+ if e.response.StatusCode != 400 {
+ t.Errorf("Expected statuscode 400 for sequence %d got %d", e.response.Sequence, e.response.StatusCode)
+ }
+ }
+ }
+}
+
func sentEventMatcher(uuid string) func(eventshubmain.EventInfo) error {
return func(ei eventshubmain.EventInfo) error {
if (ei.Kind == eventshubmain.EventSent || ei.Kind == eventshubmain.EventResponse) && ei.SentId == uuid {
