Ported Broker ingress conformance test to the new test framework (#4859)

* Ported Broker ingress conformance test to the new test framework Moved broker features under features/broker Signed-off-by: Francesco Guardiani <francescoguard@gmail.com> * Copyright Signed-off-by: Francesco Guardiani <francescoguard@gmail.com> * Fix rebase issues Signed-off-by: Francesco Guardiani <francescoguard@gmail.com> * Simplified with new methods Signed-off-by: Francesco Guardiani <francescoguard@gmail.com> * Should this be a setup? Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>
knative · Feb 18, 2021 · 403aefa · 403aefa
1 parent c44a698
commit 403aefa
Show file tree

Hide file tree

Showing 8 changed files with 182 additions and 73 deletions.
diff --git a/test/rekt/broker_test.go b/test/rekt/broker_test.go
@@ -26,7 +26,7 @@ import (
 	"knative.dev/reconciler-test/pkg/k8s"
 	"knative.dev/reconciler-test/pkg/knative"
 
-	"knative.dev/eventing/test/rekt/features"
+	"knative.dev/eventing/test/rekt/features/broker"
 )
 
 // TestBrokerAsMiddleware
@@ -41,10 +41,27 @@ func TestBrokerAsMiddleware(t *testing.T) {
 	)
 
 	// Install and wait for a Ready Broker.
-	env.Prerequisite(ctx, t, features.BrokerGoesReady("default", "MTChannelBroker"))
+	env.Prerequisite(ctx, t, broker.BrokerGoesReady("default", "MTChannelBroker"))
 
 	// Test that a Broker can act as middleware.
-	env.Test(ctx, t, features.BrokerAsMiddleware("default"))
+	env.Test(ctx, t, broker.SourceToSink("default"))
+
+	env.Finish()
+}
+
+func TestBrokerIngressConformance(t *testing.T) {
+	t.Parallel()
+
+	ctx, env := global.Environment(
+		knative.WithKnativeNamespace(system.Namespace()),
+		knative.WithLoggingConfig,
+		knative.WithTracingConfig,
+		k8s.WithEventListener,
+	)
+
+	for _, f := range broker.BrokerIngressConformanceFeatures("MTChannelBroker") {
+		env.Test(ctx, t, f)
+	}
 
 	env.Finish()
 }
diff --git a/test/rekt/features/broker/ingress.go b/test/rekt/features/broker/ingress.go
@@ -0,0 +1,116 @@
+/*
+Copyright 2021 The Knative Authors
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package broker
+
+import (
+	cloudevents "github.com/cloudevents/sdk-go/v2"
+	cetest "github.com/cloudevents/sdk-go/v2/test"
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+
+	"knative.dev/eventing/test/rekt/features"
+	"knative.dev/eventing/test/rekt/resources/broker"
+	"knative.dev/eventing/test/rekt/resources/svc"
+	"knative.dev/eventing/test/rekt/resources/trigger"
+
+	. "github.com/cloudevents/sdk-go/v2/test"
+	. "knative.dev/reconciler-test/pkg/eventshub/assert"
+)
+
+func BrokerIngressConformanceFeatures(brokerClass string) []*feature.Feature {
+	var feats []*feature.Feature
+	for _, version := range []string{cloudevents.VersionV03, cloudevents.VersionV1} {
+		for _, enc := range []cloudevents.Encoding{cloudevents.EncodingBinary, cloudevents.EncodingStructured} {
+			feats = append(feats, brokerIngressConformanceFeature(brokerClass, version, enc))
+		}
+	}
+	feats = append(feats, brokerIngressConformanceBadEvent(brokerClass))
+	return feats
+}
+
+func brokerIngressConformanceFeature(brokerClass string, version string, enc cloudevents.Encoding) *feature.Feature {
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	event := cetest.FullEvent()
+	event.SetSpecVersion(version)
+
+	f := new(feature.Feature)
+	f.Name = "BrokerIngress" + version + enc.String()
+
+	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
+	f.Setup("install broker", broker.Install(brokerName, broker.WithBrokerClass(brokerClass)))
+	f.Setup("install trigger", trigger.Install(triggerName, brokerName, trigger.WithSubscriber(svc.AsRef(sinkName), "")))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName, features.Interval, features.Timeout))
+
+	f.Setup("install source", eventshub.Install(
+		sourceName,
+		eventshub.StartSenderToResource(broker.Gvr(), brokerName),
+		eventshub.InputEventWithEncoding(event, enc),
+	))
+
+	f.Stable("ingress supports v"+version).
+		Must("accept the event", OnStore(sourceName).Match(
+			MatchKind(EventResponse),
+			MatchStatusCode(202),
+		).AtLeast(1)).
+		Must("deliver the event",
+			OnStore(sinkName).MatchEvent(
+				HasId(event.ID()),
+				HasSpecVersion(event.SpecVersion()),
+			).AtLeast(1))
+
+	return f
+}
+
+func brokerIngressConformanceBadEvent(brokerClass string) *feature.Feature {
+	sourceName := feature.MakeRandomK8sName("source")
+	sinkName := feature.MakeRandomK8sName("sink")
+	triggerName := feature.MakeRandomK8sName("trigger")
+	brokerName := feature.MakeRandomK8sName("broker")
+
+	eventID := "four-hundred-on-bad-ce"
+
+	f := new(feature.Feature)
+	f.Name = "BrokerIngressConformanceBadEvent"
+
+	f.Setup("install sink", eventshub.Install(sinkName, eventshub.StartReceiver))
+	f.Setup("install broker", broker.Install(brokerName, broker.WithBrokerClass(brokerClass)))
+	f.Setup("install trigger", trigger.Install(triggerName, brokerName, trigger.WithSubscriber(svc.AsRef(sinkName), "")))
+	f.Setup("broker is addressable", broker.IsAddressable(brokerName, features.Interval, features.Timeout))
+
+	f.Setup("install source", eventshub.Install(sourceName,
+		eventshub.StartSenderToResource(broker.Gvr(), brokerName),
+		eventshub.InputHeader("ce-specversion", "9000.1"),
+		eventshub.InputHeader("ce-type", "sometype"),
+		eventshub.InputHeader("ce-source", "400.request.sender.test.knative.dev"),
+		eventshub.InputHeader("ce-id", eventID),
+		eventshub.InputBody(";la}{kjsdf;oai2095{}{}8234092349807asdfashdf"),
+	))
+
+	f.Stable("ingress").
+		Must("respond with 400 on bad event", OnStore(sourceName).Match(
+			MatchKind(EventResponse),
+			MatchStatusCode(400),
+		).AtLeast(1)).
+		Must("must not propagate bad event",
+			OnStore(sinkName).MatchEvent(HasId(eventID)).Not())
+
+	return f
+}
diff --git a/test/rekt/features/trigger_feature.go → test/rekt/features/broker/readyness.go b/test/rekt/features/trigger_feature.go → test/rekt/features/broker/readyness.go
@@ -14,14 +14,16 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package features
+package broker
 
 import (
 	"fmt"
 
 	"knative.dev/reconciler-test/pkg/feature"
 
+	"knative.dev/eventing/test/rekt/features"
 	"knative.dev/eventing/test/rekt/resources/broker"
+	brokercfg "knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/svc"
 	"knative.dev/eventing/test/rekt/resources/trigger"
 )
@@ -44,10 +46,31 @@ func TriggerGoesReady(name, brokerName string) *feature.Feature {
 	f.Setup(fmt.Sprintf("install trigger %q", name), trigger.Install(name, brokerName, cfg...))
 
 	// Wait for a ready broker.
-	f.Requirement("broker is ready", broker.IsReady(brokerName, interval, timeout))
+	f.Requirement("broker is ready", broker.IsReady(brokerName, features.Interval, features.Timeout))
 
 	f.Stable("trigger").
-		Must("be ready", trigger.IsReady(name, interval, timeout))
+		Must("be ready", trigger.IsReady(name, features.Interval, features.Timeout))
+
+	return f
+}
+
+// BrokerGoesReady returns a feature that will create a Broker of the given
+// name and class, and confirm it becomes ready with an address.
+func BrokerGoesReady(name, class string) *feature.Feature {
+	cfg := []brokercfg.CfgFn(nil)
+
+	f := new(feature.Feature)
+
+	// Set the class of the broker.
+	if class != "" {
+		cfg = append(cfg, brokercfg.WithBrokerClass(class))
+	}
+
+	f.Setup(fmt.Sprintf("install broker %q", name), broker.Install(name, cfg...))
+
+	f.Stable("broker").
+		Must("be ready", broker.IsReady(name, features.Interval, features.Timeout)).
+		Must("be addressable", broker.IsAddressable(name, features.Interval, features.Timeout))
 
 	return f
 }
diff --git a/test/rekt/features/middleware_feature.go → test/rekt/features/broker/source_to_sink.go b/test/rekt/features/middleware_feature.go → test/rekt/features/broker/source_to_sink.go
@@ -14,25 +14,27 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package features
+package broker
 
 import (
 	"context"
 	"testing"
 
+	"knative.dev/reconciler-test/pkg/eventshub"
+	"knative.dev/reconciler-test/pkg/feature"
+
+	"knative.dev/eventing/test/rekt/features"
 	"knative.dev/eventing/test/rekt/resources/broker"
 	"knative.dev/eventing/test/rekt/resources/svc"
 	"knative.dev/eventing/test/rekt/resources/trigger"
-	"knative.dev/reconciler-test/pkg/eventshub"
-	"knative.dev/reconciler-test/pkg/feature"
 
 	. "github.com/cloudevents/sdk-go/v2/test"
 	. "knative.dev/reconciler-test/pkg/eventshub/assert"
 )
 
-// BrokerAsMiddleware tests to see if a Ready Broker acts as middleware.
+// SourceToSink tests to see if a Ready Broker acts as middleware.
 // LoadGenerator --> in [Broker] out --> Recorder
-func BrokerAsMiddleware(brokerName string) *feature.Feature {
+func SourceToSink(brokerName string) *feature.Feature {
 	source := feature.MakeRandomK8sName("source")
 	sink := feature.MakeRandomK8sName("sink")
 	via := feature.MakeRandomK8sName("via")
@@ -48,10 +50,10 @@ func BrokerAsMiddleware(brokerName string) *feature.Feature {
 	// Install the trigger
 	f.Setup("install trigger", trigger.Install(via, brokerName, cfg...))
 
-	f.Setup("trigger goes ready", trigger.IsReady(via, interval, timeout))
+	f.Setup("trigger goes ready", trigger.IsReady(via, features.Interval, features.Timeout))
 
 	f.Setup("install source", func(ctx context.Context, t *testing.T) {
-		u, err := broker.Address(ctx, brokerName, interval, timeout)
+		u, err := broker.Address(ctx, brokerName, features.Interval, features.Timeout)
 		if err != nil || u == nil {
 			t.Error("failed to get the address of the broker", brokerName, err)
 		}
@@ -60,9 +62,7 @@ func BrokerAsMiddleware(brokerName string) *feature.Feature {
 
 	f.Stable("broker as middleware").
 		Must("deliver an event",
-			func(ctx context.Context, t *testing.T) {
-				eventshub.StoreFromContext(ctx, sink).AssertExact(1, MatchEvent(HasId(event.ID())))
-			})
+			OnStore(sink).MatchEvent(HasId(event.ID())).Exact(1))
 
 	return f
 }
diff --git a/test/rekt/features/broker_feature.go b/test/rekt/features/broker_feature.go
diff --git a/test/rekt/features/defaults.go b/test/rekt/features/defaults.go
@@ -19,6 +19,6 @@ package features
 import "time"
 
 const (
-	interval = 3 * time.Second
-	timeout  = 1 * time.Minute
+	Interval = 3 * time.Second
+	Timeout  = 1 * time.Minute
 )
diff --git a/test/rekt/resources/broker/broker.go b/test/rekt/resources/broker/broker.go
@@ -36,7 +36,7 @@ import (
 
 type CfgFn func(map[string]interface{})
 
-func gvr() schema.GroupVersionResource {
+func Gvr() schema.GroupVersionResource {
 	return schema.GroupVersionResource{Group: "eventing.knative.dev", Version: "v1", Resource: "brokers"}
 }
 
@@ -109,21 +109,21 @@ func Install(name string, opts ...CfgFn) feature.StepFn {
 
 // IsReady tests to see if a Broker becomes ready within the time given.
 func IsReady(name string, interval, timeout time.Duration) feature.StepFn {
-	return k8s.IsReady(gvr(), name, interval, timeout)
+	return k8s.IsReady(Gvr(), name, interval, timeout)
 }
 
 // IsAddressable tests to see if a Broker becomes addressable within the  time
 // given.
 func IsAddressable(name string, interval, timeout time.Duration) feature.StepFn {
-	return k8s.IsAddressable(gvr(), name, interval, timeout)
+	return k8s.IsAddressable(Gvr(), name, interval, timeout)
 }
 
 // Address returns a broker's address.
 func Address(ctx context.Context, name string, interval, timeout time.Duration) (*apis.URL, error) {
 	var addr *apis.URL
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		var err error
-		addr, err = k8s.Address(ctx, gvr(), name)
+		addr, err = k8s.Address(ctx, Gvr(), name)
 		if err == nil && addr == nil {
 			// keep polling
 			return false, nil

diff --git a/test/rekt/smoke_test.go b/test/rekt/smoke_test.go
@@ -23,7 +23,7 @@ import (
 
 	_ "knative.dev/pkg/system/testing"
 
-	"knative.dev/eventing/test/rekt/features"
+	"knative.dev/eventing/test/rekt/features/broker"
 )
 
 // TestSmoke_Broker
@@ -41,7 +41,7 @@ func TestSmoke_Broker(t *testing.T) {
 	}
 
 	for _, name := range names {
-		env.Test(ctx, t, features.BrokerGoesReady(name, "MTChannelBroker"))
+		env.Test(ctx, t, broker.BrokerGoesReady(name, "MTChannelBroker"))
 	}
 
 	env.Finish()
@@ -62,10 +62,10 @@ func TestSmoke_Trigger(t *testing.T) {
 	}
 	brokerName := "broker-rekt"
 
-	env.Prerequisite(ctx, t, features.BrokerGoesReady(brokerName, "MTChannelBroker"))
+	env.Prerequisite(ctx, t, broker.BrokerGoesReady(brokerName, "MTChannelBroker"))
 
 	for _, name := range names {
-		env.Test(ctx, t, features.TriggerGoesReady(name, brokerName))
+		env.Test(ctx, t, broker.TriggerGoesReady(name, brokerName))
 	}
 
 	env.Finish()