Add e2e test for KafkaChannel

test/base/constants.go

@@ -0,0 +1,59 @@
+/*
+Copyright 2019 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 base
+
+const (
+	// InMemoryProvisioner is the in-memory provisioner, which is also the default one.
+	InMemoryProvisioner = "in-memory"
+	// GCPPubSubProvisioner is the gcp-pubsub provisioner, which is under contrib/gcppubsub.
+	GCPPubSubProvisioner = "gcp-pubsub"
+	// KafkaProvisioner is the kafka provisioner, which is under contrib/kafka.
+	KafkaProvisioner = "kafka"
+	// NatssProvisioner is the natss provisioner, which is under contrib/natss
+	NatssProvisioner = "natss"
+)
+
+// API versions for the resources.
+const (
+	EventingAPIVersion  = "eventing.knative.dev/v1alpha1"
+	SourcesAPIVersion   = "sources.eventing.knative.dev/v1alpha1"
+	MessagingAPIVersion = "messaging.knative.dev/v1alpha1"
+)
+
+// Kind for eventing resources.
+const (
+	ChannelKind                   string = "Channel"
+	SubscriptionKind              string = "Subscription"
+	ClusterChannelProvisionerKind string = "ClusterChannelProvisioner"
+
+	BrokerKind  string = "Broker"
+	TriggerKind string = "Trigger"
+)
+
+// Kind for messaging resources.
+const (
+	InMemoryChannelKind string = "InMemoryChannel"
+	KafkaChannelKind    string = "KafkaChannel"
+	NatssChannelKind    string = "NatssChannel"
+)
+
+// Kind for sources resources.
+const (
+	CronJobSourceKind   string = "CronJobSource"
+	ContainerSourceKind string = "ContainerSource"
+	ApiServerSourceKind string = "ApiServerSource"
+)

test/base/generics.go

@@ -21,8 +21,11 @@ import (
 	"github.com/knative/pkg/apis/duck"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
 )
 
 // MetaResource includes necessary meta data to retrieve the generic Kubernetes resource.
@@ -31,37 +34,66 @@ type MetaResource struct {
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 }
 
-// MetaEventing returns a MetaResource that can represent a Knative Eventing resource.
-func MetaEventing(name, namespace, kind string) *MetaResource {
-	return Meta(name, namespace, kind, eventingAPIVersion)
+// MetaResourceList includes necessary meta data to retrieve the generic Kubernetes resource list.
+type MetaResourceList struct {
+	metav1.TypeMeta `json:",inline"`
+	Namespace       string
 }
 
-// MetaSource returns a MetaResource that can represent a Knative Sources resource.
-func MetaSource(name, namespace, kind string) *MetaResource {
-	return Meta(name, namespace, kind, sourcesAPIVersion)
-}
-
-// Meta returns a MetaResource built from the given name, namespace and kind.
-func Meta(name, namespace, kind, apiVersion string) *MetaResource {
+// NewMetaResource returns a MetaResource built from the given name, namespace and typemeta.
+func NewMetaResource(name, namespace string, typemeta *metav1.TypeMeta) *MetaResource {
 	return &MetaResource{
+		TypeMeta: *typemeta,
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: namespace,
 			Name:      name,
 		},
-		TypeMeta: metav1.TypeMeta{
-			Kind:       kind,
-			APIVersion: apiVersion,
-		},
+	}
+}
+
+// NewMetaResourceList returns a MetaResourceList built from the given namespace and typemeta.
+func NewMetaResourceList(namespace string, typemeta *metav1.TypeMeta) *MetaResourceList {
+	return &MetaResourceList{
+		TypeMeta:  *typemeta,
+		Namespace: namespace,
 	}
 }
 
 // GetGenericObject returns a generic object representing a Kubernetes resource.
 // Callers can cast this returned object to other objects that implement the corresponding duck-type.
-func GetGenericObject(dynamicClient dynamic.Interface, obj *MetaResource, rtype apis.Listable) (runtime.Object, error) {
-	// get the resource's name, namespace and gvr
-	name := obj.Name
-	namespace := obj.Namespace
-	gvk := obj.GroupVersionKind()
+func GetGenericObject(
+	dynamicClient dynamic.Interface,
+	obj *MetaResource,
+	rtype apis.Listable,
+) (runtime.Object, error) {
+	lister, err := getGenericLister(dynamicClient, obj.GroupVersionKind(), obj.Namespace, rtype)
+	if err != nil {
+		return nil, err
+	}
+	return lister.Get(obj.Name)
+}
+
+// GetGenericObjectList returns a generic object list representing a list of Kubernetes resource.
+func GetGenericObjectList(
+	dynamicClient dynamic.Interface,
+	objList *MetaResourceList,
+	rtype apis.Listable,
+) ([]runtime.Object, error) {
+	lister, err := getGenericLister(dynamicClient, objList.GroupVersionKind(), objList.Namespace, rtype)
+	if err != nil {
+		return nil, err
+	}
+	return lister.List(labels.Everything())
+}
+
+// getGenericLister returns a GenericNamespacedLister, which can be used to get resources in the namespace.
+func getGenericLister(
+	dynamicClient dynamic.Interface,
+	gvk schema.GroupVersionKind,
+	namespace string,
+	rtype apis.Listable,
+) (cache.GenericNamespaceLister, error) {
+	// get the resource's namespace and gvr
 	gvr, _ := meta.UnsafeGuessKindToResource(gvk)
 
 	stopChannel := make(chan struct{})
@@ -73,5 +105,5 @@ func GetGenericObject(dynamicClient dynamic.Interface, obj *MetaResource, rtype
 	if err != nil {
 		return nil, err
 	}
-	return lister.ByNamespace(namespace).Get(name)
+	return lister.ByNamespace(namespace), nil
 }

test/base/resource_checks.go

@@ -28,6 +28,7 @@ import (
 	duckv1beta1 "github.com/knative/pkg/apis/duck/v1beta1"
 	"go.opencensus.io/trace"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/dynamic"
 )
@@ -44,18 +45,35 @@ const (
 // name the metric that is emitted to track how long it took for
 // the resource to get into the state checked by isResourceReady.
 func WaitForResourceReady(dynamicClient dynamic.Interface, obj *MetaResource) error {
-	metricName := fmt.Sprintf("WaitForResourceReady/%s", obj.Name)
+	metricName := fmt.Sprintf("WaitForResourceReady/%s/%s", obj.Namespace, obj.Name)
 	_, span := trace.StartSpan(context.Background(), metricName)
 	defer span.End()
 
 	return wait.PollImmediate(interval, timeout, func() (bool, error) {
-		return isResourceReady(dynamicClient, obj)
+		untyped, err := GetGenericObject(dynamicClient, obj, &duckv1beta1.KResource{})
+		return isResourceReady(untyped, err)
+	})
+}
+
+// WaitForResourcesReady waits until all the specified resources in the given namespace are ready.
+func WaitForResourcesReady(dynamicClient dynamic.Interface, objList *MetaResourceList) error {
+	metricName := fmt.Sprintf("WaitForResourcesReady/%s", objList.Namespace)
+	_, span := trace.StartSpan(context.Background(), metricName)
+	defer span.End()
+
+	return wait.PollImmediate(interval, timeout, func() (bool, error) {
+		untypeds, err := GetGenericObjectList(dynamicClient, objList, &duckv1beta1.KResource{})
+		for _, untyped := range untypeds {
+			if isReady, err := isResourceReady(untyped, err); !isReady {
+				return isReady, err
+			}
+		}
+		return true, nil
 	})
 }
 
 // isResourceReady leverage duck-type to check if the given MetaResource is in ready state
-func isResourceReady(dynamicClient dynamic.Interface, obj *MetaResource) (bool, error) {
-	untyped, err := GetGenericObject(dynamicClient, obj, &duckv1beta1.KResource{})
+func isResourceReady(obj runtime.Object, err error) (bool, error) {
 	if k8serrors.IsNotFound(err) {
 		// Return false as we are not done yet.
 		// We swallow the error to keep on polling.
@@ -66,6 +84,6 @@ func isResourceReady(dynamicClient dynamic.Interface, obj *MetaResource) (bool,
 		return false, err
 	}
 
-	kr := untyped.(*duckv1beta1.KResource)
+	kr := obj.(*duckv1beta1.KResource)
 	return kr.Status.GetCondition(apis.ConditionReady).IsTrue(), nil
 }