Add e2e test for api server source

test/base/resources/kube.go

@@ -111,6 +111,23 @@ func EventTransformationPod(name string, event *CloudEvent) *corev1.Pod {
 	}
 }
 
+// HelloWorldPod creates a Pod that logs "Hello, World!".
+func HelloWorldPod(name string) *corev1.Pod {
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            "helloworld",
+				Image:           pkgTest.ImagePath("helloworld"),
+				ImagePullPolicy: corev1.PullAlways,
+			}},
+			RestartPolicy: corev1.RestartPolicyNever,
+		},
+	}
+}
+
 // Service creates a Kubernetes Service with the given name, namespace, and
 // selector. Port 8080 is set as the target port.
 func Service(name string, selector map[string]string) *corev1.Service {
@@ -160,3 +177,19 @@ func ClusterRoleBinding(saName, crName, namespace string) *rbacv1.ClusterRoleBin
 		},
 	}
 }
+
+// EventWatcherClusterRole creates a Kubernetes ClusterRole that can be used to watch Events.
+func EventWatcherClusterRole(crName string) *rbacv1.ClusterRole {
+	return &rbacv1.ClusterRole{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: crName,
+		},
+		Rules: []rbacv1.PolicyRule{
+			rbacv1.PolicyRule{
+				APIGroups: []string{rbacv1.APIGroupAll},
+				Resources: []string{"events"},
+				Verbs:     []string{"get", "list", "watch"},
+			},
+		},
+	}
+}

test/base/resources/sources.go

@@ -31,6 +31,9 @@ type CronJobSourceOption func(*sourcesv1alpha1.CronJobSource)
 // ContainerSourceOption enables further configuration of a ContainerSource.
 type ContainerSourceOption func(*sourcesv1alpha1.ContainerSource)
 
+// ApiServerSourceOption enables further configuration of an ApiServerSource.
+type ApiServerSourceOption func(*sourcesv1alpha1.ApiServerSource)
+
 // WithSinkServiceForCronJobSource returns an option that adds a Kubernetes Service sink for the given CronJobSource.
 func WithSinkServiceForCronJobSource(name string) CronJobSourceOption {
 	return func(cjs *sourcesv1alpha1.CronJobSource) {
@@ -133,3 +136,39 @@ func ContainerSourceBasicTemplate(
 	}
 	return podTemplateSpec
 }
+
+// WithServiceAccountForApiServerSource returns an option that adds a ServiceAccount for the given ApiServerSource.
+func WithServiceAccountForApiServerSource(saName string) ApiServerSourceOption {
+	return func(apiServerSource *sourcesv1alpha1.ApiServerSource) {
+		apiServerSource.Spec.ServiceAccountName = saName
+	}
+}
+
+// WithSinkServiceForApiServerSource returns an option that adds a Kubernetes Service sink for the given ApiServerSource.
+func WithSinkServiceForApiServerSource(name string) ApiServerSourceOption {
+	return func(apiServerSource *sourcesv1alpha1.ApiServerSource) {
+		apiServerSource.Spec.Sink = pkgTest.CoreV1ObjectReference(ServiceKind, CoreAPIVersion, name)
+	}
+}
+
+// ApiServerSource returns an ApiServer EventSource.
+func ApiServerSource(
+	name string,
+	resources []sourcesv1alpha1.ApiServerResource,
+	mode string,
+	options ...ApiServerSourceOption,
+) *sourcesv1alpha1.ApiServerSource {
+	apiServerSource := &sourcesv1alpha1.ApiServerSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: sourcesv1alpha1.ApiServerSourceSpec{
+			Resources: resources,
+			Mode:      mode,
+		},
+	}
+	for _, option := range options {
+		option(apiServerSource)
+	}
+	return apiServerSource
+}

test/common/creation.go

@@ -18,6 +18,7 @@ package common
 
 import (
 	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
+	sourcesv1alpha1 "github.com/knative/eventing/pkg/apis/sources/v1alpha1"
 	"github.com/knative/eventing/test/base/resources"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
@@ -183,6 +184,25 @@ func (client *Client) CreateContainerSourceOrFail(
 	client.Cleaner.AddObj(containerSource)
 }
 
+// CreateApiServerSourceOrFail will create an ApiServerSource
+func (client *Client) CreateApiServerSourceOrFail(
+	name string,
+	apiServerSourceResources []sourcesv1alpha1.ApiServerResource,
+	mode string,
+	options ...resources.ApiServerSourceOption,
+) {
+	namespace := client.Namespace
+	apiServerSource := resources.ApiServerSource(name, apiServerSourceResources, mode, options...)
+
+	apiServerSources := client.Eventing.SourcesV1alpha1().ApiServerSources(namespace)
+	// update apiServerSource with the new reference
+	apiServerSource, err := apiServerSources.Create(apiServerSource)
+	if err != nil {
+		client.T.Fatalf("Failed to create apiserversource %q: %v", name, err)
+	}
+	client.Cleaner.AddObj(apiServerSource)
+}
+
 // WithService returns an option that creates a Service binded with the given pod.
 func WithService(name string) func(*corev1.Pod, *Client) error {
 	return func(pod *corev1.Pod, client *Client) error {
@@ -233,3 +253,12 @@ func (client *Client) CreateServiceAccountAndBindingOrFail(saName, crName string
 	}
 	client.Cleaner.Add(rbacAPIGroup, rbacAPIVersion, "clusterrolebindings", "", crb.GetName())
 }
+
+// CreateClusterRoleOrFail creates the given ClusterRole or fail the test if there is an error.
+func (client *Client) CreateClusterRoleOrFail(cr *rbacv1.ClusterRole) {
+	crs := client.Kube.Kube.RbacV1().ClusterRoles()
+	if _, err := crs.Create(cr); err != nil {
+		client.T.Fatalf("Failed to create cluster role %q: %v", cr.Name, err)
+	}
+	client.Cleaner.Add(rbacAPIGroup, rbacAPIVersion, "clusterroles", "", cr.Name)
+}

test/common/operation.go

@@ -113,6 +113,7 @@ func (client *Client) WaitForAllTestResourcesReady() error {
 		KafkaChannelTypeMeta,
 		InMemoryChannelTypeMeta,
 		NatssChannelTypeMeta,
+		ApiServerSourceTypeMeta,
 		CronJobSourceTypeMeta,
 		ContainerSourceTypeMeta,
 	}

test/common/typemeta.go

@@ -47,6 +47,9 @@ var CronJobSourceTypeMeta = SourcesTypeMeta(resources.CronJobSourceKind)
 // ContainerSourceTypeMeta is the TypeMeta ref for ContainerSource.
 var ContainerSourceTypeMeta = SourcesTypeMeta(resources.ContainerSourceKind)
 
+// ApiServerSourceTypeMeta is the TypeMeta ref for ApiServerSource.
+var ApiServerSourceTypeMeta = SourcesTypeMeta(resources.ApiServerSourceKind)
+
 // SourcesTypeMeta returns the TypeMeta ref for an eventing sources resource.
 func SourcesTypeMeta(kind string) *metav1.TypeMeta {
 	return &metav1.TypeMeta{

test/e2e/source_api_server_test.go

@@ -0,0 +1,83 @@
+// +build e2e
+
+/*
+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 e2e
+
+import (
+	"testing"
+
+	sourcesv1alpha1 "github.com/knative/eventing/pkg/apis/sources/v1alpha1"
+	"github.com/knative/eventing/test/base/resources"
+	"github.com/knative/eventing/test/common"
+)
+
+func TestApiServerSource(t *testing.T) {
+	const (
+		apiServerSourceName = "e2e-api-server-source"
+
+		clusterRoleName    = "event-watcher-cr"
+		serviceAccountName = "event-watcher-sa"
+		helloworldPodName  = "e2e-api-server-source-helloworld-pod"
+		loggerPodName      = "e2e-api-server-source-logger-pod"
+	)
+
+	client := Setup(t, true)
+	defer TearDown(client)
+
+	// creates ServiceAccount and ClusterRoleBinding with default cluster-admin role
+	cr := resources.EventWatcherClusterRole(clusterRoleName)
+	client.CreateClusterRoleOrFail(cr)
+	client.CreateServiceAccountAndBindingOrFail(serviceAccountName, clusterRoleName)
+
+	// create event logger pod and service
+	loggerPod := resources.EventLoggerPod(loggerPodName)
+	client.CreatePodOrFail(loggerPod, common.WithService(loggerPodName))
+
+	// create the ApiServerSource
+	// apiServerSourceResources is the list of resources to watch for this ApiServerSource
+	apiServerSourceResources := []sourcesv1alpha1.ApiServerResource{
+		sourcesv1alpha1.ApiServerResource{
+			APIVersion: "v1",
+			Kind:       "Event",
+		},
+	}
+	// mode is the watch mode: `Ref` sends only the reference to the resource, `Resource` sends the full resource.
+	mode := "Ref"
+	client.CreateApiServerSourceOrFail(
+		apiServerSourceName,
+		apiServerSourceResources,
+		mode,
+		resources.WithServiceAccountForApiServerSource(serviceAccountName),
+		resources.WithSinkServiceForApiServerSource(loggerPodName),
+	)
+
+	// wait for all test resources to be ready
+	if err := client.WaitForAllTestResourcesReady(); err != nil {
+		t.Fatalf("Failed to get all test resources ready: %v", err)
+	}
+
+	helloworldPod := resources.HelloWorldPod(helloworldPodName)
+	client.CreatePodOrFail(helloworldPod)
+
+	// verify the logger service receives the event(s)
+	// TODO(Fredy-Z): right now it's only doing a very basic check by looking for the "Event" word,
+	//                we can add a json matcher to improve it in the future.
+	data := "Event"
+	if err := client.CheckLog(loggerPodName, common.CheckerContains(data)); err != nil {
+		t.Fatalf("String %q does not appear in logs of logger pod %q: %v", data, loggerPodName, err)
+	}
+}

test/test_images/helloworld/main.go

@@ -0,0 +1,23 @@
+/*
+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 main
+
+import "log"
+
+func main() {
+	log.Println("Hello, World!")
+}