diff --git a/test/base/resource_checks.go b/test/base/resource_checks.go
index c5cd459fbbf..c52f0c58940 100644
--- a/test/base/resource_checks.go
+++ b/test/base/resource_checks.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// resource_checks.go contains functions which check resources until they
+// This file contains functions which check resources until they
 // get into the state desired by the caller or time out.
 
 package base
@@ -24,6 +24,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/pkg/apis"
 	duckv1beta1 "github.com/knative/pkg/apis/duck/v1beta1"
 	"go.opencensus.io/trace"
@@ -44,7 +45,7 @@ const (
 // it is done, returns an error or timeout. desc will be used to
 // 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 {
+func WaitForResourceReady(dynamicClient dynamic.Interface, obj *resources.MetaResource) error {
 	metricName := fmt.Sprintf("WaitForResourceReady/%s/%s", obj.Namespace, obj.Name)
 	_, span := trace.StartSpan(context.Background(), metricName)
 	defer span.End()
@@ -56,7 +57,7 @@ func WaitForResourceReady(dynamicClient dynamic.Interface, obj *MetaResource) er
 }
 
 // WaitForResourcesReady waits until all the specified resources in the given namespace are ready.
-func WaitForResourcesReady(dynamicClient dynamic.Interface, objList *MetaResourceList) error {
+func WaitForResourcesReady(dynamicClient dynamic.Interface, objList *resources.MetaResourceList) error {
 	metricName := fmt.Sprintf("WaitForResourcesReady/%s", objList.Namespace)
 	_, span := trace.StartSpan(context.Background(), metricName)
 	defer span.End()
diff --git a/test/base/generics.go b/test/base/resource_getters.go
similarity index 70%
rename from test/base/generics.go
rename to test/base/resource_getters.go
index 40dcf205521..caef2f6f43e 100644
--- a/test/base/generics.go
+++ b/test/base/resource_getters.go
@@ -14,13 +14,15 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
+// This file contains functions which get actual resources given the meta resource.
+
 package base
 
 import (
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/pkg/apis"
 	"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"
@@ -28,42 +30,11 @@ import (
 	"k8s.io/client-go/tools/cache"
 )
 
-// MetaResource includes necessary meta data to retrieve the generic Kubernetes resource.
-type MetaResource struct {
-	metav1.TypeMeta   `json:",inline"`
-	metav1.ObjectMeta `json:"metadata,omitempty"`
-}
-
-// MetaResourceList includes necessary meta data to retrieve the generic Kubernetes resource list.
-type MetaResourceList struct {
-	metav1.TypeMeta `json:",inline"`
-	Namespace       string
-}
-
-// 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,
-		},
-	}
-}
-
-// 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,
+	obj *resources.MetaResource,
 	rtype apis.Listable,
 ) (runtime.Object, error) {
 	lister, err := getGenericLister(dynamicClient, obj.GroupVersionKind(), obj.Namespace, rtype)
@@ -76,7 +47,7 @@ func GetGenericObject(
 // GetGenericObjectList returns a generic object list representing a list of Kubernetes resource.
 func GetGenericObjectList(
 	dynamicClient dynamic.Interface,
-	objList *MetaResourceList,
+	objList *resources.MetaResourceList,
 	rtype apis.Listable,
 ) ([]runtime.Object, error) {
 	lister, err := getGenericLister(dynamicClient, objList.GroupVersionKind(), objList.Namespace, rtype)
diff --git a/test/base/resource_inspectors.go b/test/base/resource_inspectors.go
index 791a8bebca6..a330f5a2e66 100644
--- a/test/base/resource_inspectors.go
+++ b/test/base/resource_inspectors.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-// resource_inspectors.go contains functions which get property values for
+// This file contains functions which get property values for
 // resources provided by the caller.
 
 package base
@@ -22,12 +22,13 @@ package base
 import (
 	"fmt"
 
+	"github.com/knative/eventing/test/base/resources"
 	duckv1alpha1 "github.com/knative/pkg/apis/duck/v1alpha1"
 	"k8s.io/client-go/dynamic"
 )
 
 // GetAddressableURI returns the uri for the given resource that implements Addressable duck-type.
-func GetAddressableURI(dynamicClient dynamic.Interface, obj *MetaResource) (string, error) {
+func GetAddressableURI(dynamicClient dynamic.Interface, obj *resources.MetaResource) (string, error) {
 	untyped, err := GetGenericObject(dynamicClient, obj, &duckv1alpha1.AddressableType{})
 	if err != nil {
 		return "", err
diff --git a/test/base/resources.go b/test/base/resources.go
deleted file mode 100644
index 2fbe6d37c0d..00000000000
--- a/test/base/resources.go
+++ /dev/null
@@ -1,445 +0,0 @@
-/*
-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
-
-// resources contains functions that construct Eventing CRs and other Kubernetes resources.
-// TODO(Fredy-Z): break this file into multiple files when it grows too large.
-
-import (
-	"fmt"
-
-	kafkamessagingv1alpha1 "github.com/knative/eventing/contrib/kafka/pkg/apis/messaging/v1alpha1"
-	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
-	sourcesv1alpha1 "github.com/knative/eventing/pkg/apis/sources/v1alpha1"
-	pkgTest "github.com/knative/pkg/test"
-	corev1 "k8s.io/api/core/v1"
-	rbacv1 "k8s.io/api/rbac/v1"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/util/intstr"
-	"k8s.io/apimachinery/pkg/util/uuid"
-)
-
-// TriggerOption enables further configuration of a Trigger.
-type TriggerOption func(*eventingv1alpha1.Trigger)
-
-// SubscriptionOption enables further configuration of a Subscription.
-type SubscriptionOption func(*eventingv1alpha1.Subscription)
-
-// CronJobSourceOption enables further configuration of a CronJobSource.
-type CronJobSourceOption func(*sourcesv1alpha1.CronJobSource)
-
-// ContainerSourceOption enables further configuration of a ContainerSource.
-type ContainerSourceOption func(*sourcesv1alpha1.ContainerSource)
-
-// clusterChannelProvisioner returns a ClusterChannelProvisioner for a given name.
-func clusterChannelProvisioner(name string) *corev1.ObjectReference {
-	if name == "" {
-		return nil
-	}
-	return pkgTest.CoreV1ObjectReference(ClusterChannelProvisionerKind, EventingAPIVersion, name)
-}
-
-// channelRef returns an ObjectReference for a given Channel Name.
-func channelRef(name string, typemeta *metav1.TypeMeta) *corev1.ObjectReference {
-	return pkgTest.CoreV1ObjectReference(typemeta.Kind, typemeta.APIVersion, name)
-}
-
-// Channel returns a Channel with the specified provisioner.
-func Channel(name, provisioner string) *eventingv1alpha1.Channel {
-	return &eventingv1alpha1.Channel{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: eventingv1alpha1.ChannelSpec{
-			Provisioner: clusterChannelProvisioner(provisioner),
-		},
-	}
-}
-
-// KafkaChannel returns a KafkaChannel.
-func KafkaChannel(name string) *kafkamessagingv1alpha1.KafkaChannel {
-	return &kafkamessagingv1alpha1.KafkaChannel{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-	}
-}
-
-// WithSubscriberForSubscription returns an option that adds a Subscriber for the given Subscription.
-func WithSubscriberForSubscription(name string) SubscriptionOption {
-	return func(s *eventingv1alpha1.Subscription) {
-		if name != "" {
-			s.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
-				Ref: pkgTest.CoreV1ObjectReference("Service", "v1", name),
-			}
-		}
-	}
-}
-
-// WithReply returns an options that adds a ReplyStrategy for the given Subscription.
-func WithReply(name string, typemeta *metav1.TypeMeta) SubscriptionOption {
-	return func(s *eventingv1alpha1.Subscription) {
-		if name != "" {
-			s.Spec.Reply = &eventingv1alpha1.ReplyStrategy{
-				Channel: pkgTest.CoreV1ObjectReference(typemeta.Kind, typemeta.APIVersion, name),
-			}
-		}
-	}
-}
-
-// Subscription returns a Subscription.
-func Subscription(
-	name, channelName string,
-	channelTypeMeta *metav1.TypeMeta,
-	options ...SubscriptionOption,
-) *eventingv1alpha1.Subscription {
-	subscription := &eventingv1alpha1.Subscription{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: eventingv1alpha1.SubscriptionSpec{
-			Channel: *channelRef(channelName, channelTypeMeta),
-		},
-	}
-	for _, option := range options {
-		option(subscription)
-	}
-	return subscription
-}
-
-// Broker returns a Broker.
-func Broker(name, provisioner string) *eventingv1alpha1.Broker {
-	return &eventingv1alpha1.Broker{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: eventingv1alpha1.BrokerSpec{
-			ChannelTemplate: &eventingv1alpha1.ChannelSpec{
-				Provisioner: clusterChannelProvisioner(provisioner),
-			},
-		},
-	}
-}
-
-// WithTriggerFilter returns an option that adds a TriggerFilter for the given Trigger.
-func WithTriggerFilter(eventSource, eventType string) TriggerOption {
-	return func(t *eventingv1alpha1.Trigger) {
-		triggerFilter := &eventingv1alpha1.TriggerFilter{
-			SourceAndType: &eventingv1alpha1.TriggerFilterSourceAndType{
-				Type:   eventType,
-				Source: eventSource,
-			},
-		}
-		t.Spec.Filter = triggerFilter
-	}
-}
-
-// WithBroker returns an option that adds a Broker for the given Trigger.
-func WithBroker(brokerName string) TriggerOption {
-	return func(t *eventingv1alpha1.Trigger) {
-		t.Spec.Broker = brokerName
-	}
-}
-
-// WithSubscriberRefForTrigger returns an option that adds a Subscriber Ref for the given Trigger.
-func WithSubscriberRefForTrigger(name string) TriggerOption {
-	return func(t *eventingv1alpha1.Trigger) {
-		if name != "" {
-			t.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
-				Ref: pkgTest.CoreV1ObjectReference("Service", "v1", name),
-			}
-		}
-	}
-}
-
-// WithSubscriberURIForTrigger returns an option that adds a Subscriber URI for the given Trigger.
-func WithSubscriberURIForTrigger(uri string) TriggerOption {
-	return func(t *eventingv1alpha1.Trigger) {
-		t.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
-			URI: &uri,
-		}
-	}
-}
-
-// Trigger returns a Trigger.
-func Trigger(name string, options ...TriggerOption) *eventingv1alpha1.Trigger {
-	trigger := &eventingv1alpha1.Trigger{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-	}
-	for _, option := range options {
-		option(trigger)
-	}
-	return trigger
-}
-
-// WithSinkServiceForCronJobSource returns an option that adds a Kubernetes Service sink for the given CronJobSource.
-func WithSinkServiceForCronJobSource(name string) CronJobSourceOption {
-	return func(cjs *sourcesv1alpha1.CronJobSource) {
-		cjs.Spec.Sink = pkgTest.CoreV1ObjectReference("Service", "v1", name)
-	}
-}
-
-// WithServiceAccountForCronJobSource returns an option that adds a ServiceAccount for the given CronJobSource.
-func WithServiceAccountForCronJobSource(saName string) CronJobSourceOption {
-	return func(cjs *sourcesv1alpha1.CronJobSource) {
-		cjs.Spec.ServiceAccountName = saName
-	}
-}
-
-// CronJobSource returns a CronJob EventSource.
-func CronJobSource(
-	name,
-	schedule,
-	data string,
-	options ...CronJobSourceOption,
-) *sourcesv1alpha1.CronJobSource {
-	cronJobSource := &sourcesv1alpha1.CronJobSource{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: sourcesv1alpha1.CronJobSourceSpec{
-			Schedule: schedule,
-			Data:     data,
-		},
-	}
-	for _, option := range options {
-		option(cronJobSource)
-	}
-	return cronJobSource
-}
-
-// WithTemplateForContainerSource returns an option that adds a template for the given ContainerSource.
-func WithTemplateForContainerSource(template *corev1.PodTemplateSpec) ContainerSourceOption {
-	return func(cs *sourcesv1alpha1.ContainerSource) {
-		cs.Spec.Template = template
-	}
-}
-
-// WithSinkServiceForContainerSource returns an option that adds a Kubernetes Service sink for the given ContainerSource.
-func WithSinkServiceForContainerSource(name string) ContainerSourceOption {
-	return func(cs *sourcesv1alpha1.ContainerSource) {
-		cs.Spec.Sink = pkgTest.CoreV1ObjectReference("Service", "v1", name)
-	}
-}
-
-// ContainerSource returns a Container EventSource.
-func ContainerSource(
-	name string,
-	options ...ContainerSourceOption,
-) *sourcesv1alpha1.ContainerSource {
-	containerSource := &sourcesv1alpha1.ContainerSource{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-	}
-	for _, option := range options {
-		option(containerSource)
-	}
-	return containerSource
-}
-
-// ContainerSourceBasicTemplate returns a basic template that can be used in ContainerSource.
-func ContainerSourceBasicTemplate(
-	name,
-	namespace,
-	imageName string,
-	args []string,
-) *corev1.PodTemplateSpec {
-	envVars := []corev1.EnvVar{
-		{
-			Name:  "POD_NAME",
-			Value: name,
-		},
-		{
-			Name:  "POD_NAMESPACE",
-			Value: namespace,
-		},
-	}
-
-	podTemplateSpec := &corev1.PodTemplateSpec{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{
-				{
-					Name:            imageName,
-					Image:           pkgTest.ImagePath(imageName),
-					ImagePullPolicy: corev1.PullAlways,
-					Args:            args,
-					Env:             envVars,
-				},
-			},
-		},
-	}
-	return podTemplateSpec
-}
-
-// CloudEvent specifies the arguments for a CloudEvent sent by the sendevent
-// binary.
-type CloudEvent struct {
-	ID       string
-	Type     string
-	Source   string
-	Data     string
-	Encoding string // binary or structured
-}
-
-// TypeAndSource specifies the type and source of an Event.
-type TypeAndSource struct {
-	Type   string
-	Source string
-}
-
-// CloudEvent related constants.
-const (
-	CloudEventEncodingBinary     = "binary"
-	CloudEventEncodingStructured = "structured"
-	CloudEventDefaultEncoding    = CloudEventEncodingBinary
-	CloudEventDefaultType        = "dev.knative.test.event"
-)
-
-// EventSenderPod creates a Pod that sends a single event to the given address.
-func EventSenderPod(name string, sink string, event *CloudEvent) *corev1.Pod {
-	if event.Encoding == "" {
-		event.Encoding = CloudEventEncodingBinary
-	}
-	return &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{
-				Name:            "sendevent",
-				Image:           pkgTest.ImagePath("sendevent"),
-				ImagePullPolicy: corev1.PullAlways,
-				Args: []string{
-					"-event-id",
-					event.ID,
-					"-event-type",
-					event.Type,
-					"-source",
-					event.Source,
-					"-data",
-					event.Data,
-					"-encoding",
-					event.Encoding,
-					"-sink",
-					sink,
-				},
-			}},
-			//TODO restart on failure?
-			RestartPolicy: corev1.RestartPolicyNever,
-		},
-	}
-}
-
-// EventLoggerPod creates a Pod that logs events received.
-func EventLoggerPod(name string) *corev1.Pod {
-	return &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:   name,
-			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{
-				Name:            "logevents",
-				Image:           pkgTest.ImagePath("logevents"),
-				ImagePullPolicy: corev1.PullAlways,
-			}},
-			RestartPolicy: corev1.RestartPolicyAlways,
-		},
-	}
-}
-
-// EventTransformationPod creates a Pod that transforms events received.
-func EventTransformationPod(name string, event *CloudEvent) *corev1.Pod {
-	return &corev1.Pod{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:   name,
-			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
-		},
-		Spec: corev1.PodSpec{
-			Containers: []corev1.Container{{
-				Name:            "transformevents",
-				Image:           pkgTest.ImagePath("transformevents"),
-				ImagePullPolicy: corev1.PullAlways,
-				Args: []string{
-					"-event-type",
-					event.Type,
-					"-event-source",
-					event.Source,
-					"-event-data",
-					event.Data,
-				},
-			}},
-			RestartPolicy: corev1.RestartPolicyAlways,
-		},
-	}
-}
-
-// Service creates a Kubernetes Service with the given name, namespace, and
-// selector. Port 8080 is assumed the target port.
-func Service(name string, selector map[string]string) *corev1.Service {
-	return &corev1.Service{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: name,
-		},
-		Spec: corev1.ServiceSpec{
-			Selector: selector,
-			Ports: []corev1.ServicePort{{
-				Name:       "http",
-				Port:       80,
-				Protocol:   corev1.ProtocolTCP,
-				TargetPort: intstr.FromInt(8080),
-			}},
-		},
-	}
-}
-
-// ServiceAccount creates a Kubernetes ServiceAccount with the given name and namespace.
-func ServiceAccount(name, namespace string) *corev1.ServiceAccount {
-	return &corev1.ServiceAccount{
-		ObjectMeta: metav1.ObjectMeta{
-			Name:      name,
-			Namespace: namespace,
-		},
-	}
-}
-
-// ClusterRoleBinding creates a Kubernetes ClusterRoleBinding with the given ServiceAccount name, ClusterRole name and namespace.
-func ClusterRoleBinding(saName, crName, namespace string) *rbacv1.ClusterRoleBinding {
-	return &rbacv1.ClusterRoleBinding{
-		ObjectMeta: metav1.ObjectMeta{
-			Name: fmt.Sprintf("%s-%s-admin", saName, namespace),
-		},
-		Subjects: []rbacv1.Subject{
-			{
-				Kind:      "ServiceAccount",
-				Name:      saName,
-				Namespace: namespace,
-			},
-		},
-		RoleRef: rbacv1.RoleRef{
-			Kind:     "ClusterRole",
-			Name:     crName,
-			APIGroup: rbacv1.SchemeGroupVersion.Group,
-		},
-	}
-}
diff --git a/test/base/resources/cloud_event.go b/test/base/resources/cloud_event.go
new file mode 100644
index 00000000000..6f89df99420
--- /dev/null
+++ b/test/base/resources/cloud_event.go
@@ -0,0 +1,35 @@
+/*
+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 resources
+
+// CloudEvent specifies the arguments for a CloudEvent sent by the sendevent
+// binary.
+type CloudEvent struct {
+	ID       string
+	Type     string
+	Source   string
+	Data     string
+	Encoding string // binary or structured
+}
+
+// CloudEvent related constants.
+const (
+	CloudEventEncodingBinary     = "binary"
+	CloudEventEncodingStructured = "structured"
+	CloudEventDefaultEncoding    = CloudEventEncodingBinary
+	CloudEventDefaultType        = "dev.knative.test.event"
+)
diff --git a/test/base/constants.go b/test/base/resources/constants.go
similarity index 93%
rename from test/base/constants.go
rename to test/base/resources/constants.go
index fde6bbd37b2..ca065ac924e 100644
--- a/test/base/constants.go
+++ b/test/base/resources/constants.go
@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package base
+package resources
 
 const (
 	// InMemoryProvisioner is the in-memory provisioner, which is also the default one.
@@ -29,11 +29,17 @@ const (
 
 // API versions for the resources.
 const (
+	CoreAPIVersion      = "v1"
 	EventingAPIVersion  = "eventing.knative.dev/v1alpha1"
 	SourcesAPIVersion   = "sources.eventing.knative.dev/v1alpha1"
 	MessagingAPIVersion = "messaging.knative.dev/v1alpha1"
 )
 
+// Kind for core Kubernetes resources.
+const (
+	ServiceKind string = "Service"
+)
+
 // Kind for eventing resources.
 const (
 	ChannelKind                   string = "Channel"
diff --git a/test/base/resources/eventing.go b/test/base/resources/eventing.go
new file mode 100644
index 00000000000..726dae4085b
--- /dev/null
+++ b/test/base/resources/eventing.go
@@ -0,0 +1,166 @@
+/*
+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 resources
+
+// This file contains functions that construct Eventing resources.
+
+import (
+	eventingv1alpha1 "github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
+	pkgTest "github.com/knative/pkg/test"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// TriggerOption enables further configuration of a Trigger.
+type TriggerOption func(*eventingv1alpha1.Trigger)
+
+// SubscriptionOption enables further configuration of a Subscription.
+type SubscriptionOption func(*eventingv1alpha1.Subscription)
+
+// clusterChannelProvisioner returns a ClusterChannelProvisioner for a given name.
+func clusterChannelProvisioner(name string) *corev1.ObjectReference {
+	if name == "" {
+		return nil
+	}
+	return pkgTest.CoreV1ObjectReference(ClusterChannelProvisionerKind, EventingAPIVersion, name)
+}
+
+// channelRef returns an ObjectReference for a given Channel Name.
+func channelRef(name string, typemeta *metav1.TypeMeta) *corev1.ObjectReference {
+	return pkgTest.CoreV1ObjectReference(typemeta.Kind, typemeta.APIVersion, name)
+}
+
+// Channel returns a Channel with the specified provisioner.
+func Channel(name, provisioner string) *eventingv1alpha1.Channel {
+	return &eventingv1alpha1.Channel{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: eventingv1alpha1.ChannelSpec{
+			Provisioner: clusterChannelProvisioner(provisioner),
+		},
+	}
+}
+
+// WithSubscriberForSubscription returns an option that adds a Subscriber for the given Subscription.
+func WithSubscriberForSubscription(name string) SubscriptionOption {
+	return func(s *eventingv1alpha1.Subscription) {
+		if name != "" {
+			s.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
+				Ref: pkgTest.CoreV1ObjectReference(ServiceKind, CoreAPIVersion, name),
+			}
+		}
+	}
+}
+
+// WithReply returns an options that adds a ReplyStrategy for the given Subscription.
+func WithReply(name string, typemeta *metav1.TypeMeta) SubscriptionOption {
+	return func(s *eventingv1alpha1.Subscription) {
+		if name != "" {
+			s.Spec.Reply = &eventingv1alpha1.ReplyStrategy{
+				Channel: pkgTest.CoreV1ObjectReference(typemeta.Kind, typemeta.APIVersion, name),
+			}
+		}
+	}
+}
+
+// Subscription returns a Subscription.
+func Subscription(
+	name, channelName string,
+	channelTypeMeta *metav1.TypeMeta,
+	options ...SubscriptionOption,
+) *eventingv1alpha1.Subscription {
+	subscription := &eventingv1alpha1.Subscription{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: eventingv1alpha1.SubscriptionSpec{
+			Channel: *channelRef(channelName, channelTypeMeta),
+		},
+	}
+	for _, option := range options {
+		option(subscription)
+	}
+	return subscription
+}
+
+// Broker returns a Broker.
+func Broker(name, provisioner string) *eventingv1alpha1.Broker {
+	return &eventingv1alpha1.Broker{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: eventingv1alpha1.BrokerSpec{
+			ChannelTemplate: &eventingv1alpha1.ChannelSpec{
+				Provisioner: clusterChannelProvisioner(provisioner),
+			},
+		},
+	}
+}
+
+// WithTriggerFilter returns an option that adds a TriggerFilter for the given Trigger.
+func WithTriggerFilter(eventSource, eventType string) TriggerOption {
+	return func(t *eventingv1alpha1.Trigger) {
+		triggerFilter := &eventingv1alpha1.TriggerFilter{
+			SourceAndType: &eventingv1alpha1.TriggerFilterSourceAndType{
+				Type:   eventType,
+				Source: eventSource,
+			},
+		}
+		t.Spec.Filter = triggerFilter
+	}
+}
+
+// WithBroker returns an option that adds a Broker for the given Trigger.
+func WithBroker(brokerName string) TriggerOption {
+	return func(t *eventingv1alpha1.Trigger) {
+		t.Spec.Broker = brokerName
+	}
+}
+
+// WithSubscriberRefForTrigger returns an option that adds a Subscriber Ref for the given Trigger.
+func WithSubscriberRefForTrigger(name string) TriggerOption {
+	return func(t *eventingv1alpha1.Trigger) {
+		if name != "" {
+			t.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
+				Ref: pkgTest.CoreV1ObjectReference(ServiceKind, CoreAPIVersion, name),
+			}
+		}
+	}
+}
+
+// WithSubscriberURIForTrigger returns an option that adds a Subscriber URI for the given Trigger.
+func WithSubscriberURIForTrigger(uri string) TriggerOption {
+	return func(t *eventingv1alpha1.Trigger) {
+		t.Spec.Subscriber = &eventingv1alpha1.SubscriberSpec{
+			URI: &uri,
+		}
+	}
+}
+
+// Trigger returns a Trigger.
+func Trigger(name string, options ...TriggerOption) *eventingv1alpha1.Trigger {
+	trigger := &eventingv1alpha1.Trigger{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	for _, option := range options {
+		option(trigger)
+	}
+	return trigger
+}
diff --git a/test/base/resources/generics.go b/test/base/resources/generics.go
new file mode 100644
index 00000000000..6059612fe7b
--- /dev/null
+++ b/test/base/resources/generics.go
@@ -0,0 +1,52 @@
+/*
+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 resources
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// MetaResource includes necessary meta data to retrieve the generic Kubernetes resource.
+type MetaResource struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+}
+
+// MetaResourceList includes necessary meta data to retrieve the generic Kubernetes resource list.
+type MetaResourceList struct {
+	metav1.TypeMeta `json:",inline"`
+	Namespace       string
+}
+
+// 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,
+		},
+	}
+}
+
+// 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,
+	}
+}
diff --git a/test/base/resources/kube.go b/test/base/resources/kube.go
new file mode 100644
index 00000000000..6dd1dee63db
--- /dev/null
+++ b/test/base/resources/kube.go
@@ -0,0 +1,162 @@
+/*
+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 resources
+
+// This file contains functions that construct common Kubernetes resources.
+
+import (
+	"fmt"
+
+	pkgTest "github.com/knative/pkg/test"
+	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	"k8s.io/apimachinery/pkg/util/uuid"
+)
+
+// EventSenderPod creates a Pod that sends a single event to the given address.
+func EventSenderPod(name string, sink string, event *CloudEvent) *corev1.Pod {
+	const imageName = "sendevent"
+	if event.Encoding == "" {
+		event.Encoding = CloudEventEncodingBinary
+	}
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            imageName,
+				Image:           pkgTest.ImagePath(imageName),
+				ImagePullPolicy: corev1.PullAlways,
+				Args: []string{
+					"-event-id",
+					event.ID,
+					"-event-type",
+					event.Type,
+					"-source",
+					event.Source,
+					"-data",
+					event.Data,
+					"-encoding",
+					event.Encoding,
+					"-sink",
+					sink,
+				},
+			}},
+			//TODO restart on failure?
+			RestartPolicy: corev1.RestartPolicyNever,
+		},
+	}
+}
+
+// EventLoggerPod creates a Pod that logs events received.
+func EventLoggerPod(name string) *corev1.Pod {
+	const imageName = "logevents"
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            imageName,
+				Image:           pkgTest.ImagePath(imageName),
+				ImagePullPolicy: corev1.PullAlways,
+			}},
+			RestartPolicy: corev1.RestartPolicyAlways,
+		},
+	}
+}
+
+// EventTransformationPod creates a Pod that transforms events received.
+func EventTransformationPod(name string, event *CloudEvent) *corev1.Pod {
+	const imageName = "transformevents"
+	return &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:   name,
+			Labels: map[string]string{"e2etest": string(uuid.NewUUID())},
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{{
+				Name:            imageName,
+				Image:           pkgTest.ImagePath(imageName),
+				ImagePullPolicy: corev1.PullAlways,
+				Args: []string{
+					"-event-type",
+					event.Type,
+					"-event-source",
+					event.Source,
+					"-event-data",
+					event.Data,
+				},
+			}},
+			RestartPolicy: corev1.RestartPolicyAlways,
+		},
+	}
+}
+
+// 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 {
+	return &corev1.Service{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: corev1.ServiceSpec{
+			Selector: selector,
+			Ports: []corev1.ServicePort{{
+				Name:       "http",
+				Port:       80,
+				Protocol:   corev1.ProtocolTCP,
+				TargetPort: intstr.FromInt(8080),
+			}},
+		},
+	}
+}
+
+// ServiceAccount creates a Kubernetes ServiceAccount with the given name and namespace.
+func ServiceAccount(name, namespace string) *corev1.ServiceAccount {
+	return &corev1.ServiceAccount{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:      name,
+			Namespace: namespace,
+		},
+	}
+}
+
+// ClusterRoleBinding creates a Kubernetes ClusterRoleBinding with the given ServiceAccount name, ClusterRole name and namespace.
+func ClusterRoleBinding(saName, crName, namespace string) *rbacv1.ClusterRoleBinding {
+	return &rbacv1.ClusterRoleBinding{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: fmt.Sprintf("%s-%s-admin", saName, namespace),
+		},
+		Subjects: []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Name:      saName,
+				Namespace: namespace,
+			},
+		},
+		RoleRef: rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     crName,
+			APIGroup: rbacv1.SchemeGroupVersion.Group,
+		},
+	}
+}
diff --git a/test/base/resources/messaging.go b/test/base/resources/messaging.go
new file mode 100644
index 00000000000..4f673598f1d
--- /dev/null
+++ b/test/base/resources/messaging.go
@@ -0,0 +1,43 @@
+/*
+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 resources
+
+// This file contains functions that construct Messaging resources.
+
+import (
+	kafkamessagingv1alpha1 "github.com/knative/eventing/contrib/kafka/pkg/apis/messaging/v1alpha1"
+	messagingv1alpha1 "github.com/knative/eventing/pkg/apis/messaging/v1alpha1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// KafkaChannel returns a KafkaChannel resource.
+func KafkaChannel(name string) *kafkamessagingv1alpha1.KafkaChannel {
+	return &kafkamessagingv1alpha1.KafkaChannel{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+}
+
+// InMemoryChannel returns an InMemoryChannel resource.
+func InMemoryChannel(name string) *messagingv1alpha1.InMemoryChannel {
+	return &messagingv1alpha1.InMemoryChannel{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+}
diff --git a/test/base/resources/sources.go b/test/base/resources/sources.go
new file mode 100644
index 00000000000..5e8c886ede0
--- /dev/null
+++ b/test/base/resources/sources.go
@@ -0,0 +1,135 @@
+/*
+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 resources
+
+// This file contains functions that construct Sources resources.
+
+import (
+	sourcesv1alpha1 "github.com/knative/eventing/pkg/apis/sources/v1alpha1"
+	pkgTest "github.com/knative/pkg/test"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// CronJobSourceOption enables further configuration of a CronJobSource.
+type CronJobSourceOption func(*sourcesv1alpha1.CronJobSource)
+
+// ContainerSourceOption enables further configuration of a ContainerSource.
+type ContainerSourceOption func(*sourcesv1alpha1.ContainerSource)
+
+// WithSinkServiceForCronJobSource returns an option that adds a Kubernetes Service sink for the given CronJobSource.
+func WithSinkServiceForCronJobSource(name string) CronJobSourceOption {
+	return func(cjs *sourcesv1alpha1.CronJobSource) {
+		cjs.Spec.Sink = pkgTest.CoreV1ObjectReference(ServiceKind, CoreAPIVersion, name)
+	}
+}
+
+// WithServiceAccountForCronJobSource returns an option that adds a ServiceAccount for the given CronJobSource.
+func WithServiceAccountForCronJobSource(saName string) CronJobSourceOption {
+	return func(cjs *sourcesv1alpha1.CronJobSource) {
+		cjs.Spec.ServiceAccountName = saName
+	}
+}
+
+// CronJobSource returns a CronJob EventSource.
+func CronJobSource(
+	name,
+	schedule,
+	data string,
+	options ...CronJobSourceOption,
+) *sourcesv1alpha1.CronJobSource {
+	cronJobSource := &sourcesv1alpha1.CronJobSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: sourcesv1alpha1.CronJobSourceSpec{
+			Schedule: schedule,
+			Data:     data,
+		},
+	}
+	for _, option := range options {
+		option(cronJobSource)
+	}
+	return cronJobSource
+}
+
+// WithTemplateForContainerSource returns an option that adds a template for the given ContainerSource.
+func WithTemplateForContainerSource(template *corev1.PodTemplateSpec) ContainerSourceOption {
+	return func(cs *sourcesv1alpha1.ContainerSource) {
+		cs.Spec.Template = template
+	}
+}
+
+// WithSinkServiceForContainerSource returns an option that adds a Kubernetes Service sink for the given ContainerSource.
+func WithSinkServiceForContainerSource(name string) ContainerSourceOption {
+	return func(cs *sourcesv1alpha1.ContainerSource) {
+		cs.Spec.Sink = pkgTest.CoreV1ObjectReference(ServiceKind, CoreAPIVersion, name)
+	}
+}
+
+// ContainerSource returns a Container EventSource.
+func ContainerSource(
+	name string,
+	options ...ContainerSourceOption,
+) *sourcesv1alpha1.ContainerSource {
+	containerSource := &sourcesv1alpha1.ContainerSource{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+	}
+	for _, option := range options {
+		option(containerSource)
+	}
+	return containerSource
+}
+
+// ContainerSourceBasicTemplate returns a basic template that can be used in ContainerSource.
+func ContainerSourceBasicTemplate(
+	name,
+	namespace,
+	imageName string,
+	args []string,
+) *corev1.PodTemplateSpec {
+	envVars := []corev1.EnvVar{
+		corev1.EnvVar{
+			Name:  "POD_NAME",
+			Value: name,
+		},
+		corev1.EnvVar{
+			Name:  "POD_NAMESPACE",
+			Value: namespace,
+		},
+	}
+
+	podTemplateSpec := &corev1.PodTemplateSpec{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: name,
+		},
+		Spec: corev1.PodSpec{
+			Containers: []corev1.Container{
+				corev1.Container{
+					Name:            imageName,
+					Image:           pkgTest.ImagePath(imageName),
+					ImagePullPolicy: corev1.PullAlways,
+					Args:            args,
+					Env:             envVars,
+				},
+			},
+		},
+	}
+	return podTemplateSpec
+}
diff --git a/test/common/config.go b/test/common/config.go
index 96b5ecaac14..cf7082adfb3 100644
--- a/test/common/config.go
+++ b/test/common/config.go
@@ -17,38 +17,43 @@ limitations under the License.
 package common
 
 import (
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // DefaultClusterChannelProvisioner is the default ClusterChannelProvisioner we will run tests against.
-const DefaultClusterChannelProvisioner = base.InMemoryProvisioner
+const DefaultClusterChannelProvisioner = resources.InMemoryProvisioner
 
 // ValidProvisionersMap saves the provisioner-features mapping.
 // Each pair means the provisioner support the list of features.
 var ValidProvisionersMap = map[string]ChannelConfig{
-	base.InMemoryProvisioner: {
-		Features: []Feature{FeatureBasic},
+	resources.InMemoryProvisioner: ChannelConfig{
+		Features:     []Feature{FeatureBasic},
+		CRDSupported: true,
 	},
-	base.GCPPubSubProvisioner: {
+	resources.GCPPubSubProvisioner: ChannelConfig{
 		Features: []Feature{FeatureBasic, FeatureRedelivery, FeaturePersistence},
 	},
-	base.KafkaProvisioner: {
+	resources.KafkaProvisioner: ChannelConfig{
 		Features:     []Feature{FeatureBasic, FeatureRedelivery, FeaturePersistence},
 		CRDSupported: true,
 	},
-	base.NatssProvisioner: {
+	resources.NatssProvisioner: ChannelConfig{
 		Features: []Feature{FeatureBasic, FeatureRedelivery, FeaturePersistence},
 	},
 }
 
+// ChannelConfig includes general configuration for a Channel provisioner.
 type ChannelConfig struct {
 	Features     []Feature
 	CRDSupported bool
 }
 
-var OperatorChannelMap = map[string]*metav1.TypeMeta{
-	base.KafkaProvisioner: KafkaChannelTypeMeta,
+// ProvisionerChannelMap saves the mapping between provisioners and CRD channel typemeta.
+// TODO(Fredy-Z): this map will not be needed anymore when we delete the provisioner implementation.
+var ProvisionerChannelMap = map[string]*metav1.TypeMeta{
+	resources.KafkaProvisioner:    KafkaChannelTypeMeta,
+	resources.InMemoryProvisioner: InMemoryChannelTypeMeta,
 }
 
 // Feature is the feature supported by the Channel provisioner.
diff --git a/test/common/creation.go b/test/common/creation.go
index ca27eb37401..9852edcf62c 100644
--- a/test/common/creation.go
+++ b/test/common/creation.go
@@ -17,7 +17,7 @@ limitations under the License.
 package common
 
 import (
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	corev1 "k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -31,21 +31,27 @@ var rbacAPIGroup = rbacv1.SchemeGroupVersion.Group
 var rbacAPIVersion = rbacv1.SchemeGroupVersion.Version
 
 // CreateChannelOrFail will create a Channel Resource in Eventing.
-// TODO(Fredy-Z): This is a workaround when there are both provisioner and Channel CRDs in this repo.
-//                isCRD needs to be deleted when the provisioner implementation is removed.
 func (client *Client) CreateChannelOrFail(name string, channelTypeMeta *metav1.TypeMeta, provisionerName string) {
 	namespace := client.Namespace
 	switch channelTypeMeta.Kind {
-	case base.ChannelKind:
-		channel := base.Channel(name, provisionerName)
+	case resources.ChannelKind:
+		channel := resources.Channel(name, provisionerName)
 		channels := client.Eventing.EventingV1alpha1().Channels(namespace)
 		channel, err := channels.Create(channel)
 		if err != nil {
 			client.T.Fatalf("Failed to create channel %q: %v", name, err)
 		}
 		client.Cleaner.AddObj(channel)
-	case base.KafkaChannelKind:
-		channel := base.KafkaChannel(name)
+	case resources.InMemoryChannelKind:
+		channel := resources.InMemoryChannel(name)
+		channels := client.Eventing.MessagingV1alpha1().InMemoryChannels(namespace)
+		channel, err := channels.Create(channel)
+		if err != nil {
+			client.T.Fatalf("Failed to create %q %q: %v", channelTypeMeta.Kind, name, err)
+		}
+		client.Cleaner.AddObj(channel)
+	case resources.KafkaChannelKind:
+		channel := resources.KafkaChannel(name)
 		channels := client.Kafka.MessagingV1alpha1().KafkaChannels(namespace)
 		channel, err := channels.Create(channel)
 		if err != nil {
@@ -66,10 +72,10 @@ func (client *Client) CreateChannelsOrFail(names []string, channelTypeMeta *meta
 func (client *Client) CreateSubscriptionOrFail(
 	name, channelName string,
 	channelTypeMeta *metav1.TypeMeta,
-	options ...base.SubscriptionOption,
+	options ...resources.SubscriptionOption,
 ) {
 	namespace := client.Namespace
-	subscription := base.Subscription(name, channelName, channelTypeMeta, options...)
+	subscription := resources.Subscription(name, channelName, channelTypeMeta, options...)
 
 	subscriptions := client.Eventing.EventingV1alpha1().Subscriptions(namespace)
 	// update subscription with the new reference
@@ -85,7 +91,7 @@ func (client *Client) CreateSubscriptionsOrFail(
 	names []string,
 	channelName string,
 	channelTypeMeta *metav1.TypeMeta,
-	options ...base.SubscriptionOption,
+	options ...resources.SubscriptionOption,
 ) {
 	for _, name := range names {
 		client.CreateSubscriptionOrFail(name, channelName, channelTypeMeta, options...)
@@ -95,7 +101,7 @@ func (client *Client) CreateSubscriptionsOrFail(
 // CreateBrokerOrFail will create a Broker or fail the test if there is an error.
 func (client *Client) CreateBrokerOrFail(name, provisionerName string) {
 	namespace := client.Namespace
-	broker := base.Broker(name, provisionerName)
+	broker := resources.Broker(name, provisionerName)
 
 	brokers := client.Eventing.EventingV1alpha1().Brokers(namespace)
 	// update broker with the new reference
@@ -114,9 +120,9 @@ func (client *Client) CreateBrokersOrFail(names []string, provisionerName string
 }
 
 // CreateTriggerOrFail will create a Trigger or fail the test if there is an error.
-func (client *Client) CreateTriggerOrFail(name string, options ...base.TriggerOption) {
+func (client *Client) CreateTriggerOrFail(name string, options ...resources.TriggerOption) {
 	namespace := client.Namespace
-	trigger := base.Trigger(name, options...)
+	trigger := resources.Trigger(name, options...)
 
 	triggers := client.Eventing.EventingV1alpha1().Triggers(namespace)
 	// update trigger with the new reference
@@ -132,10 +138,10 @@ func (client *Client) CreateCronJobSourceOrFail(
 	name,
 	schedule,
 	data string,
-	options ...base.CronJobSourceOption,
+	options ...resources.CronJobSourceOption,
 ) {
 	namespace := client.Namespace
-	cronJobSource := base.CronJobSource(name, schedule, data, options...)
+	cronJobSource := resources.CronJobSource(name, schedule, data, options...)
 
 	cronJobSources := client.Eventing.SourcesV1alpha1().CronJobSources(namespace)
 	// update cronJobSource with the new reference
@@ -146,13 +152,13 @@ func (client *Client) CreateCronJobSourceOrFail(
 	client.Cleaner.AddObj(cronJobSource)
 }
 
-// CreateContainerSourceOrFail will create a ContainerSource.
+// CreateContainerSourceOrFail will create a ContainerSource or fail the test if there is an error.
 func (client *Client) CreateContainerSourceOrFail(
 	name string,
-	options ...base.ContainerSourceOption,
+	options ...resources.ContainerSourceOption,
 ) {
 	namespace := client.Namespace
-	containerSource := base.ContainerSource(name, options...)
+	containerSource := resources.ContainerSource(name, options...)
 
 	containerSources := client.Eventing.SourcesV1alpha1().ContainerSources(namespace)
 	// update containerSource with the new reference
@@ -167,7 +173,7 @@ func (client *Client) CreateContainerSourceOrFail(
 func WithService(name string) func(*corev1.Pod, *Client) error {
 	return func(pod *corev1.Pod, client *Client) error {
 		namespace := pod.Namespace
-		svc := base.Service(name, pod.Labels)
+		svc := resources.Service(name, pod.Labels)
 
 		svcs := client.Kube.Kube.CoreV1().Services(namespace)
 		if _, err := svcs.Create(svc); err != nil {
@@ -199,14 +205,14 @@ func (client *Client) CreatePodOrFail(pod *corev1.Pod, options ...func(*corev1.P
 // cluster-admin role.
 func (client *Client) CreateServiceAccountAndBindingOrFail(saName, crName string) {
 	namespace := client.Namespace
-	sa := base.ServiceAccount(saName, namespace)
+	sa := resources.ServiceAccount(saName, namespace)
 	sas := client.Kube.Kube.CoreV1().ServiceAccounts(namespace)
 	if _, err := sas.Create(sa); err != nil {
 		client.T.Fatalf("Failed to create service account %q: %v", saName, err)
 	}
 	client.Cleaner.Add(coreAPIGroup, coreAPIVersion, "serviceaccounts", namespace, saName)
 
-	crb := base.ClusterRoleBinding(saName, crName, namespace)
+	crb := resources.ClusterRoleBinding(saName, crName, namespace)
 	crbs := client.Kube.Kube.RbacV1().ClusterRoleBindings()
 	if _, err := crbs.Create(crb); err != nil {
 		client.T.Fatalf("Failed to create cluster role binding %q: %v", crName, err)
diff --git a/test/common/operation.go b/test/common/operation.go
index 574a475d13b..eaf7b9cb56a 100644
--- a/test/common/operation.go
+++ b/test/common/operation.go
@@ -20,6 +20,7 @@ import (
 	"time"
 
 	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	pkgTest "github.com/knative/pkg/test"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
@@ -46,7 +47,7 @@ func (client *Client) SendFakeEventToAddressable(
 	senderName,
 	addressableName string,
 	typemeta *metav1.TypeMeta,
-	event *base.CloudEvent,
+	event *resources.CloudEvent,
 ) error {
 	uri, err := client.GetAddressableURI(addressableName, typemeta)
 	if err != nil {
@@ -59,7 +60,7 @@ func (client *Client) SendFakeEventToAddressable(
 // To use this function, the given resource must have implemented the Addressable duck-type.
 func (client *Client) GetAddressableURI(addressableName string, typemeta *metav1.TypeMeta) (string, error) {
 	namespace := client.Namespace
-	metaAddressable := base.NewMetaResource(addressableName, namespace, typemeta)
+	metaAddressable := resources.NewMetaResource(addressableName, namespace, typemeta)
 	return base.GetAddressableURI(client.Dynamic, metaAddressable)
 }
 
@@ -67,10 +68,10 @@ func (client *Client) GetAddressableURI(addressableName string, typemeta *metav1
 func (client *Client) sendFakeEventToAddress(
 	senderName string,
 	uri string,
-	event *base.CloudEvent,
+	event *resources.CloudEvent,
 ) error {
 	namespace := client.Namespace
-	pod := base.EventSenderPod(senderName, uri, event)
+	pod := resources.EventSenderPod(senderName, uri, event)
 	client.CreatePodOrFail(pod)
 	if err := pkgTest.WaitForPodRunning(client.Kube, senderName, namespace); err != nil {
 		return err
@@ -82,7 +83,7 @@ func (client *Client) sendFakeEventToAddress(
 // To use this function, the given resource must have implemented the Status duck-type.
 func (client *Client) WaitForResourceReady(name string, typemeta *metav1.TypeMeta) error {
 	namespace := client.Namespace
-	metaResource := base.NewMetaResource(name, namespace, typemeta)
+	metaResource := resources.NewMetaResource(name, namespace, typemeta)
 	if err := base.WaitForResourceReady(client.Dynamic, metaResource); err != nil {
 		return err
 	}
@@ -93,7 +94,7 @@ func (client *Client) WaitForResourceReady(name string, typemeta *metav1.TypeMet
 // To use this function, the given resource must have implemented the Status duck-type.
 func (client *Client) WaitForResourcesReady(typemeta *metav1.TypeMeta) error {
 	namespace := client.Namespace
-	metaResourceList := base.NewMetaResourceList(namespace, typemeta)
+	metaResourceList := resources.NewMetaResourceList(namespace, typemeta)
 	if err := base.WaitForResourcesReady(client.Dynamic, metaResourceList); err != nil {
 		return err
 	}
@@ -110,6 +111,7 @@ func (client *Client) WaitForAllTestResourcesReady() error {
 		BrokerTypeMeta,
 		TriggerTypeMeta,
 		KafkaChannelTypeMeta,
+		InMemoryChannelTypeMeta,
 		CronJobSourceTypeMeta,
 		ContainerSourceTypeMeta,
 	}
diff --git a/test/common/typemeta.go b/test/common/typemeta.go
index db900b336bb..1754d3cce23 100644
--- a/test/common/typemeta.go
+++ b/test/common/typemeta.go
@@ -17,51 +17,54 @@ limitations under the License.
 package common
 
 import (
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
 // ChannelTypeMeta is the TypeMeta ref for Channel.
-var ChannelTypeMeta = EventingTypeMeta(base.ChannelKind)
+var ChannelTypeMeta = EventingTypeMeta(resources.ChannelKind)
 
 // SubscriptionTypeMeta is the TypeMeta ref for Subscription.
-var SubscriptionTypeMeta = EventingTypeMeta(base.SubscriptionKind)
+var SubscriptionTypeMeta = EventingTypeMeta(resources.SubscriptionKind)
 
 // BrokerTypeMeta is the TypeMeta ref for Broker.
-var BrokerTypeMeta = EventingTypeMeta(base.BrokerKind)
+var BrokerTypeMeta = EventingTypeMeta(resources.BrokerKind)
 
 // TriggerTypeMeta is the TypeMeta ref for Trigger.
-var TriggerTypeMeta = EventingTypeMeta(base.TriggerKind)
+var TriggerTypeMeta = EventingTypeMeta(resources.TriggerKind)
 
 // EventingTypeMeta returns the TypeMeta ref for an eventing resource.
 func EventingTypeMeta(kind string) *metav1.TypeMeta {
 	return &metav1.TypeMeta{
 		Kind:       kind,
-		APIVersion: base.EventingAPIVersion,
+		APIVersion: resources.EventingAPIVersion,
 	}
 }
 
 // CronJobSourceTypeMeta is the TypeMeta ref for CronJobSource.
-var CronJobSourceTypeMeta = SourcesTypeMeta(base.CronJobSourceKind)
+var CronJobSourceTypeMeta = SourcesTypeMeta(resources.CronJobSourceKind)
 
 // ContainerSourceTypeMeta is the TypeMeta ref for ContainerSource.
-var ContainerSourceTypeMeta = SourcesTypeMeta(base.ContainerSourceKind)
+var ContainerSourceTypeMeta = SourcesTypeMeta(resources.ContainerSourceKind)
 
 // SourcesTypeMeta returns the TypeMeta ref for an eventing sources resource.
 func SourcesTypeMeta(kind string) *metav1.TypeMeta {
 	return &metav1.TypeMeta{
 		Kind:       kind,
-		APIVersion: base.SourcesAPIVersion,
+		APIVersion: resources.SourcesAPIVersion,
 	}
 }
 
 // KafkaChannelTypeMeta is the TypeMeta ref for KafkaChannel.
-var KafkaChannelTypeMeta = MessagingTypeMeta(base.KafkaChannelKind)
+var KafkaChannelTypeMeta = MessagingTypeMeta(resources.KafkaChannelKind)
+
+// InMemoryChannelTypeMeta is the TypeMeta ref for InMemoryChannel.
+var InMemoryChannelTypeMeta = MessagingTypeMeta(resources.InMemoryChannelKind)
 
 // MessagingTypeMeta returns the TypeMeta ref for an eventing messaing resource.
 func MessagingTypeMeta(kind string) *metav1.TypeMeta {
 	return &metav1.TypeMeta{
 		Kind:       kind,
-		APIVersion: base.MessagingAPIVersion,
+		APIVersion: resources.MessagingAPIVersion,
 	}
 }
diff --git a/test/e2e-tests.sh b/test/e2e-tests.sh
index 6f0034a3d07..36596c80af6 100755
--- a/test/e2e-tests.sh
+++ b/test/e2e-tests.sh
@@ -33,6 +33,9 @@ readonly EVENTING_CONFIG="config/"
 # In-memory provisioner config.
 readonly IN_MEMORY_CHANNEL_CONFIG="config/provisioners/in-memory-channel/in-memory-channel.yaml"
 
+# In-memory channel CRD config.
+readonly IN_MEMORY_CHANNEL_CRD_CONFIG_DIR="config/channels/in-memory-channel"
+
 # GCP PubSub provisioner config template.
 readonly GCP_PUBSUB_CONFIG_TEMPLATE="contrib/gcppubsub/config/gcppubsub.yaml"
 # Real GCP PubSub provisioner config, generated from the template.
@@ -90,6 +93,10 @@ function test_setup() {
   ko apply -f ${IN_MEMORY_CHANNEL_CONFIG} || return 1
   wait_until_pods_running knative-eventing || fail_test "Failed to install the In-Memory ClusterChannelProvisioner"
 
+  echo "Installing In-Memory Channel CRD"
+  ko apply -f ${IN_MEMORY_CHANNEL_CRD_CONFIG_DIR} || return 1
+  wait_until_pods_running knative-eventing || fail_test "Failed to install the In-Memory Channel CRD"
+
   echo "Installing GCPPubSub ClusterChannelProvisioner"
   gcppubsub_setup || return 1
   sed "s/REPLACE_WITH_GCP_PROJECT/${E2E_PROJECT_ID}/" ${GCP_PUBSUB_CONFIG_TEMPLATE} > ${GCP_PUBSUB_CONFIG}
@@ -120,10 +127,12 @@ function test_setup() {
 
 # Tear down resources used in the eventing tests.
 function test_teardown() {
-  # Uninstall provisioners used by the tests.
   echo "Uninstalling In-Memory ClusterChannelProvisioner"
   ko delete --ignore-not-found=true --now --timeout 60s -f ${IN_MEMORY_CHANNEL_CONFIG}
 
+  echo "Uninstalling In-Memory Channel CRD"
+  ko delete --ignore-not-found=true --now --timeout 60s -f ${IN_MEMORY_CHANNEL_CRD_CONFIG_DIR}
+
   echo "Uninstalling GCPPubSub ClusterChannelProvisioner"
   gcppubsub_teardown
   ko delete --ignore-not-found=true --now --timeout 60s -f ${GCP_PUBSUB_CONFIG}
diff --git a/test/e2e/broker_channel_flow_test.go b/test/e2e/broker_channel_flow_test.go
index ea37e538095..e8d12e6f0ba 100644
--- a/test/e2e/broker_channel_flow_test.go
+++ b/test/e2e/broker_channel_flow_test.go
@@ -22,7 +22,7 @@ import (
 	"testing"
 
 	"github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -94,35 +94,35 @@ func testBrokerChannelFlow(t *testing.T, provisioner string, isCRD bool) {
 
 	// create the event we want to transform to
 	transformedEventBody := fmt.Sprintf("%s %s", eventBody, string(uuid.NewUUID()))
-	eventAfterTransformation := &base.CloudEvent{
+	eventAfterTransformation := &resources.CloudEvent{
 		Source:   eventSource2,
 		Type:     eventType2,
 		Data:     fmt.Sprintf(`{"msg":%q}`, transformedEventBody),
-		Encoding: base.CloudEventDefaultEncoding,
+		Encoding: resources.CloudEventDefaultEncoding,
 	}
 
 	// create the transformation service for trigger1
-	transformationPod := base.EventTransformationPod(transformationPodName, eventAfterTransformation)
+	transformationPod := resources.EventTransformationPod(transformationPodName, eventAfterTransformation)
 	client.CreatePodOrFail(transformationPod, common.WithService(transformationPodName))
 
 	// create trigger1 to receive the original event, and do event transformation
 	client.CreateTriggerOrFail(
 		triggerName1,
-		base.WithBroker(brokerName),
-		base.WithTriggerFilter(eventSource1, eventType1),
-		base.WithSubscriberRefForTrigger(transformationPodName),
+		resources.WithBroker(brokerName),
+		resources.WithTriggerFilter(eventSource1, eventType1),
+		resources.WithSubscriberRefForTrigger(transformationPodName),
 	)
 
 	// create logger pod and service for trigger2
-	loggerPod1 := base.EventLoggerPod(loggerPodName1)
+	loggerPod1 := resources.EventLoggerPod(loggerPodName1)
 	client.CreatePodOrFail(loggerPod1, common.WithService(loggerPodName1))
 
 	// create trigger2 to receive all the events
 	client.CreateTriggerOrFail(
 		triggerName2,
-		base.WithBroker(brokerName),
-		base.WithTriggerFilter(any, any),
-		base.WithSubscriberRefForTrigger(loggerPodName1),
+		resources.WithBroker(brokerName),
+		resources.WithTriggerFilter(any, any),
+		resources.WithSubscriberRefForTrigger(loggerPodName1),
 	)
 
 	// create channel for trigger3
@@ -137,13 +137,13 @@ func testBrokerChannelFlow(t *testing.T, provisioner string, isCRD bool) {
 	}
 	client.CreateTriggerOrFail(
 		triggerName3,
-		base.WithBroker(brokerName),
-		base.WithTriggerFilter(eventSource2, eventType2),
-		base.WithSubscriberURIForTrigger(channelURL),
+		resources.WithBroker(brokerName),
+		resources.WithTriggerFilter(eventSource2, eventType2),
+		resources.WithSubscriberURIForTrigger(channelURL),
 	)
 
 	// create logger pod and service for subscription
-	loggerPod2 := base.EventLoggerPod(loggerPodName2)
+	loggerPod2 := resources.EventLoggerPod(loggerPodName2)
 	client.CreatePodOrFail(loggerPod2, common.WithService(loggerPodName2))
 
 	// create subscription
@@ -151,7 +151,7 @@ func testBrokerChannelFlow(t *testing.T, provisioner string, isCRD bool) {
 		subscriptionName,
 		channelName,
 		channelTypeMeta,
-		base.WithSubscriberForSubscription(loggerPodName2),
+		resources.WithSubscriberForSubscription(loggerPodName2),
 	)
 
 	// wait for all test resources to be ready, so that we can start sending events
@@ -160,11 +160,11 @@ func testBrokerChannelFlow(t *testing.T, provisioner string, isCRD bool) {
 	}
 
 	// send fake CloudEvent to the broker
-	eventToSend := &base.CloudEvent{
+	eventToSend := &resources.CloudEvent{
 		Source:   eventSource1,
 		Type:     eventType1,
 		Data:     fmt.Sprintf(`{"msg":%q}`, eventBody),
-		Encoding: base.CloudEventDefaultEncoding,
+		Encoding: resources.CloudEventDefaultEncoding,
 	}
 	if err := client.SendFakeEventToAddressable(senderName, brokerName, common.BrokerTypeMeta, eventToSend); err != nil {
 		t.Fatalf("Failed to send fake CloudEvent to the broker %q", brokerName)
diff --git a/test/e2e/broker_default_test.go b/test/e2e/broker_default_test.go
index ecd6b9938db..945e330cefd 100644
--- a/test/e2e/broker_default_test.go
+++ b/test/e2e/broker_default_test.go
@@ -25,8 +25,8 @@ import (
 	"time"
 
 	"github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
-	"github.com/knative/eventing/pkg/reconciler/namespace/resources"
-	"github.com/knative/eventing/test/base"
+	pkgResources "github.com/knative/eventing/pkg/reconciler/namespace/resources"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 
 	"github.com/knative/pkg/test/logging"
@@ -37,7 +37,7 @@ const (
 	waitForFilterPodRunning = 30 * time.Second
 	selectorKey             = "end2end-test-broker-trigger"
 
-	defaultBrokerName = resources.DefaultBrokerName
+	defaultBrokerName = pkgResources.DefaultBrokerName
 	any               = v1alpha1.TriggerAnyFilter
 	eventType1        = "type1"
 	eventType2        = "type2"
@@ -45,10 +45,16 @@ const (
 	eventSource2      = "source2"
 )
 
+// eventTypeAndSource specifies the type and source of an Event.
+type eventTypeAndSource struct {
+	Type   string
+	Source string
+}
+
 // Helper struct to tie the type and sources of the events we expect to receive
 // in subscribers with the selectors we use when creating their pods.
 type eventReceiver struct {
-	typeAndSource base.TypeAndSource
+	typeAndSource eventTypeAndSource
 	selector      map[string]string
 }
 
@@ -73,16 +79,16 @@ func TestDefaultBrokerWithManyTriggers(t *testing.T) {
 	// These are the event types and sources that triggers will listen to, as well as the selectors
 	// to set  in the subscriber and services pods.
 	eventsToReceive := []eventReceiver{
-		{base.TypeAndSource{Type: any, Source: any}, newSelector()},
-		{base.TypeAndSource{Type: eventType1, Source: any}, newSelector()},
-		{base.TypeAndSource{Type: any, Source: eventSource1}, newSelector()},
-		{base.TypeAndSource{Type: eventType1, Source: eventSource1}, newSelector()},
+		{eventTypeAndSource{Type: any, Source: any}, newSelector()},
+		{eventTypeAndSource{Type: eventType1, Source: any}, newSelector()},
+		{eventTypeAndSource{Type: any, Source: eventSource1}, newSelector()},
+		{eventTypeAndSource{Type: eventType1, Source: eventSource1}, newSelector()},
 	}
 
 	// Create subscribers.
 	for _, event := range eventsToReceive {
 		subscriberName := name("dumper", event.typeAndSource.Type, event.typeAndSource.Source)
-		pod := base.EventLoggerPod(subscriberName)
+		pod := resources.EventLoggerPod(subscriberName)
 		client.CreatePodOrFail(pod, common.WithService(subscriberName))
 	}
 
@@ -91,8 +97,8 @@ func TestDefaultBrokerWithManyTriggers(t *testing.T) {
 		triggerName := name("trigger", event.typeAndSource.Type, event.typeAndSource.Source)
 		subscriberName := name("dumper", event.typeAndSource.Type, event.typeAndSource.Source)
 		client.CreateTriggerOrFail(triggerName,
-			base.WithSubscriberRefForTrigger(subscriberName),
-			base.WithTriggerFilter(event.typeAndSource.Source, event.typeAndSource.Type),
+			resources.WithSubscriberRefForTrigger(subscriberName),
+			resources.WithTriggerFilter(event.typeAndSource.Source, event.typeAndSource.Type),
 		)
 	}
 
@@ -102,7 +108,7 @@ func TestDefaultBrokerWithManyTriggers(t *testing.T) {
 	}
 
 	// These are the event types and sources that will be send.
-	eventsToSend := []base.TypeAndSource{
+	eventsToSend := []eventTypeAndSource{
 		{eventType1, eventSource1},
 		{eventType1, eventSource2},
 		{eventType2, eventSource1},
@@ -116,7 +122,7 @@ func TestDefaultBrokerWithManyTriggers(t *testing.T) {
 		// Create cloud event.
 		// Using event type and source as part of the body for easier debugging.
 		body := fmt.Sprintf("Body-%s-%s", eventToSend.Type, eventToSend.Source)
-		cloudEvent := &base.CloudEvent{
+		cloudEvent := &resources.CloudEvent{
 			Source: eventToSend.Source,
 			Type:   eventToSend.Type,
 			Data:   fmt.Sprintf(`{"msg":%q}`, body),
@@ -174,7 +180,7 @@ func newSelector() map[string]string {
 }
 
 // Checks whether we should expect to receive 'eventToSend' in 'eventReceiver' based on its type and source pattern.
-func shouldExpectEvent(eventToSend *base.TypeAndSource, receiver *eventReceiver, logf logging.FormatLogger) bool {
+func shouldExpectEvent(eventToSend *eventTypeAndSource, receiver *eventReceiver, logf logging.FormatLogger) bool {
 	if receiver.typeAndSource.Type != any && receiver.typeAndSource.Type != eventToSend.Type {
 		return false
 	}
diff --git a/test/e2e/broker_event_transformation_test.go b/test/e2e/broker_event_transformation_test.go
index 3bb2b4d028d..47514ef132b 100644
--- a/test/e2e/broker_event_transformation_test.go
+++ b/test/e2e/broker_event_transformation_test.go
@@ -22,7 +22,7 @@ import (
 	"testing"
 
 	"github.com/knative/eventing/pkg/apis/eventing/v1alpha1"
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -83,35 +83,35 @@ func testEventTransformationForTrigger(t *testing.T, provisioner string, isCRD b
 
 	// create the event we want to transform to
 	transformedEventBody := fmt.Sprintf("%s %s", eventBody, string(uuid.NewUUID()))
-	eventAfterTransformation := &base.CloudEvent{
+	eventAfterTransformation := &resources.CloudEvent{
 		Source:   eventSource2,
 		Type:     eventType2,
 		Data:     fmt.Sprintf(`{"msg":%q}`, transformedEventBody),
-		Encoding: base.CloudEventDefaultEncoding,
+		Encoding: resources.CloudEventDefaultEncoding,
 	}
 
 	// create the transformation service
-	transformationPod := base.EventTransformationPod(transformationPodName, eventAfterTransformation)
+	transformationPod := resources.EventTransformationPod(transformationPodName, eventAfterTransformation)
 	client.CreatePodOrFail(transformationPod, common.WithService(transformationPodName))
 
 	// create trigger1 for event transformation
 	client.CreateTriggerOrFail(
 		triggerName1,
-		base.WithBroker(brokerName),
-		base.WithTriggerFilter(eventSource1, eventType1),
-		base.WithSubscriberRefForTrigger(transformationPodName),
+		resources.WithBroker(brokerName),
+		resources.WithTriggerFilter(eventSource1, eventType1),
+		resources.WithSubscriberRefForTrigger(transformationPodName),
 	)
 
 	// create logger pod and service
-	loggerPod := base.EventLoggerPod(loggerPodName)
+	loggerPod := resources.EventLoggerPod(loggerPodName)
 	client.CreatePodOrFail(loggerPod, common.WithService(loggerPodName))
 
 	// create trigger2 for event receiving
 	client.CreateTriggerOrFail(
 		triggerName2,
-		base.WithBroker(brokerName),
-		base.WithTriggerFilter(eventSource2, eventType2),
-		base.WithSubscriberRefForTrigger(loggerPodName),
+		resources.WithBroker(brokerName),
+		resources.WithTriggerFilter(eventSource2, eventType2),
+		resources.WithSubscriberRefForTrigger(loggerPodName),
 	)
 
 	// wait for all test resources to be ready, so that we can start sending events
@@ -120,11 +120,11 @@ func testEventTransformationForTrigger(t *testing.T, provisioner string, isCRD b
 	}
 
 	// send fake CloudEvent to the broker
-	eventToSend := &base.CloudEvent{
+	eventToSend := &resources.CloudEvent{
 		Source:   eventSource1,
 		Type:     eventType1,
 		Data:     fmt.Sprintf(`{"msg":%q}`, eventBody),
-		Encoding: base.CloudEventDefaultEncoding,
+		Encoding: resources.CloudEventDefaultEncoding,
 	}
 	if err := client.SendFakeEventToAddressable(senderName, brokerName, common.BrokerTypeMeta, eventToSend); err != nil {
 		t.Fatalf("Failed to send fake CloudEvent to the broker %q", brokerName)
diff --git a/test/e2e/channel_chain_test.go b/test/e2e/channel_chain_test.go
index 1483a377076..94be9638e0c 100644
--- a/test/e2e/channel_chain_test.go
+++ b/test/e2e/channel_chain_test.go
@@ -22,7 +22,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -57,7 +57,7 @@ func testChannelChain(t *testing.T, provisioner string, isCRD bool) {
 	client.WaitForResourcesReady(channelTypeMeta)
 
 	// create loggerPod and expose it as a service
-	pod := base.EventLoggerPod(loggerPodName)
+	pod := resources.EventLoggerPod(loggerPodName)
 	client.CreatePodOrFail(pod, common.WithService(loggerPodName))
 
 	// create subscriptions that subscribe the first channel, and reply events directly to the second channel
@@ -65,14 +65,14 @@ func testChannelChain(t *testing.T, provisioner string, isCRD bool) {
 		subscriptionNames1,
 		channelNames[0],
 		channelTypeMeta,
-		base.WithReply(channelNames[1], channelTypeMeta),
+		resources.WithReply(channelNames[1], channelTypeMeta),
 	)
 	// create subscriptions that subscribe the second channel, and call the logging service
 	client.CreateSubscriptionsOrFail(
 		subscriptionNames2,
 		channelNames[1],
 		channelTypeMeta,
-		base.WithSubscriberForSubscription(loggerPodName),
+		resources.WithSubscriberForSubscription(loggerPodName),
 	)
 
 	// wait for all test resources to be ready, so that we can start sending events
@@ -82,11 +82,11 @@ func testChannelChain(t *testing.T, provisioner string, isCRD bool) {
 
 	// send fake CloudEvent to the first channel
 	body := fmt.Sprintf("TestChannelChainEvent %s", uuid.NewUUID())
-	event := &base.CloudEvent{
+	event := &resources.CloudEvent{
 		Source:   senderName,
-		Type:     base.CloudEventDefaultType,
+		Type:     resources.CloudEventDefaultType,
 		Data:     fmt.Sprintf(`{"msg":%q}`, body),
-		Encoding: base.CloudEventDefaultEncoding,
+		Encoding: resources.CloudEventDefaultEncoding,
 	}
 	if err := client.SendFakeEventToAddressable(senderName, channelNames[0], channelTypeMeta, event); err != nil {
 		t.Fatalf("Failed to send fake CloudEvent to the channel %q", channelNames[0])
diff --git a/test/e2e/channel_event_transformation_test.go b/test/e2e/channel_event_transformation_test.go
index bca1811a7e7..4fce17626c4 100644
--- a/test/e2e/channel_event_transformation_test.go
+++ b/test/e2e/channel_event_transformation_test.go
@@ -22,7 +22,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -59,17 +59,17 @@ func TestEventTransformationForSubscription(t *testing.T) {
 
 		// create transformation pod and service
 		transformedEventBody := fmt.Sprintf("eventBody %s", uuid.NewUUID())
-		eventAfterTransformation := &base.CloudEvent{
+		eventAfterTransformation := &resources.CloudEvent{
 			Source:   senderName,
-			Type:     base.CloudEventDefaultType,
+			Type:     resources.CloudEventDefaultType,
 			Data:     fmt.Sprintf(`{"msg":%q}`, transformedEventBody),
-			Encoding: base.CloudEventDefaultEncoding,
+			Encoding: resources.CloudEventDefaultEncoding,
 		}
-		transformationPod := base.EventTransformationPod(transformationPodName, eventAfterTransformation)
+		transformationPod := resources.EventTransformationPod(transformationPodName, eventAfterTransformation)
 		client.CreatePodOrFail(transformationPod, common.WithService(transformationPodName))
 
 		// create logger pod and service
-		loggerPod := base.EventLoggerPod(loggerPodName)
+		loggerPod := resources.EventLoggerPod(loggerPodName)
 		client.CreatePodOrFail(loggerPod, common.WithService(loggerPodName))
 
 		// create subscriptions that subscribe the first channel, use the transformation service to transform the events and then forward the transformed events to the second channel
@@ -77,15 +77,15 @@ func TestEventTransformationForSubscription(t *testing.T) {
 			subscriptionNames1,
 			channelNames[0],
 			channelTypeMeta,
-			base.WithSubscriberForSubscription(transformationPodName),
-			base.WithReply(channelNames[1], channelTypeMeta),
+			resources.WithSubscriberForSubscription(transformationPodName),
+			resources.WithReply(channelNames[1], channelTypeMeta),
 		)
 		// create subscriptions that subscribe the second channel, and forward the received events to the logger service
 		client.CreateSubscriptionsOrFail(
 			subscriptionNames2,
 			channelNames[1],
 			channelTypeMeta,
-			base.WithSubscriberForSubscription(loggerPodName),
+			resources.WithSubscriberForSubscription(loggerPodName),
 		)
 
 		// wait for all test resources to be ready, so that we can start sending events
@@ -95,11 +95,11 @@ func TestEventTransformationForSubscription(t *testing.T) {
 
 		// send fake CloudEvent to the first channel
 		eventBody := fmt.Sprintf("TestEventTransformation %s", uuid.NewUUID())
-		eventToSend := &base.CloudEvent{
+		eventToSend := &resources.CloudEvent{
 			Source:   senderName,
-			Type:     base.CloudEventDefaultType,
+			Type:     resources.CloudEventDefaultType,
 			Data:     fmt.Sprintf(`{"msg":%q}`, eventBody),
-			Encoding: base.CloudEventDefaultEncoding,
+			Encoding: resources.CloudEventDefaultEncoding,
 		}
 		if err := client.SendFakeEventToAddressable(senderName, channelNames[0], channelTypeMeta, eventToSend); err != nil {
 			st.Fatalf("Failed to send fake CloudEvent to the channel %q", channelNames[0])
diff --git a/test/e2e/channel_single_event_test.go b/test/e2e/channel_single_event_test.go
index cfa0827e647..0a4ccf36564 100644
--- a/test/e2e/channel_single_event_test.go
+++ b/test/e2e/channel_single_event_test.go
@@ -22,17 +22,17 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
 
 func TestSingleBinaryEventForChannel(t *testing.T) {
-	singleEvent(t, base.CloudEventEncodingBinary)
+	singleEvent(t, resources.CloudEventEncodingBinary)
 }
 
 func TestSingleStructuredEventForChannel(t *testing.T) {
-	singleEvent(t, base.CloudEventEncodingStructured)
+	singleEvent(t, resources.CloudEventEncodingStructured)
 }
 
 /*
@@ -57,7 +57,7 @@ func singleEvent(t *testing.T, encoding string) {
 		client.CreateChannelOrFail(channelName, channelTypeMeta, provisioner)
 
 		// create logger service as the subscriber
-		pod := base.EventLoggerPod(loggerPodName)
+		pod := resources.EventLoggerPod(loggerPodName)
 		client.CreatePodOrFail(pod, common.WithService(loggerPodName))
 
 		// create subscription to subscribe the channel, and forward the received events to the logger service
@@ -65,7 +65,7 @@ func singleEvent(t *testing.T, encoding string) {
 			subscriptionName,
 			channelName,
 			channelTypeMeta,
-			base.WithSubscriberForSubscription(loggerPodName),
+			resources.WithSubscriberForSubscription(loggerPodName),
 		)
 
 		// wait for all test resources to be ready, so that we can start sending events
@@ -75,9 +75,9 @@ func singleEvent(t *testing.T, encoding string) {
 
 		// send fake CloudEvent to the channel
 		body := fmt.Sprintf("TestSingleEvent %s", uuid.NewUUID())
-		event := &base.CloudEvent{
+		event := &resources.CloudEvent{
 			Source:   senderName,
-			Type:     base.CloudEventDefaultType,
+			Type:     resources.CloudEventDefaultType,
 			Data:     fmt.Sprintf(`{"msg":%q}`, body),
 			Encoding: encoding,
 		}
diff --git a/test/e2e/source_container_test.go b/test/e2e/source_container_test.go
index a0c48d79fda..5ba4f7762d6 100644
--- a/test/e2e/source_container_test.go
+++ b/test/e2e/source_container_test.go
@@ -21,7 +21,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -40,7 +40,7 @@ func TestContainerSource(t *testing.T) {
 	defer TearDown(client)
 
 	// create event logger pod and service
-	loggerPod := base.EventLoggerPod(loggerPodName)
+	loggerPod := resources.EventLoggerPod(loggerPodName)
 	client.CreatePodOrFail(loggerPod, common.WithService(loggerPodName))
 
 	data := fmt.Sprintf("TestContainerSource%s", uuid.NewUUID())
@@ -48,9 +48,9 @@ func TestContainerSource(t *testing.T) {
 	args := []string{"--msg=" + data}
 
 	// create container source
-	template := base.ContainerSourceBasicTemplate(templateName, client.Namespace, imageName, args)
-	templateOption := base.WithTemplateForContainerSource(template)
-	sinkOption := base.WithSinkServiceForContainerSource(loggerPodName)
+	template := resources.ContainerSourceBasicTemplate(templateName, client.Namespace, imageName, args)
+	templateOption := resources.WithTemplateForContainerSource(template)
+	sinkOption := resources.WithSinkServiceForContainerSource(loggerPodName)
 	client.CreateContainerSourceOrFail(containerSourceName, templateOption, sinkOption)
 
 	// wait for all test resources to be ready
diff --git a/test/e2e/source_cron_job_test.go b/test/e2e/source_cron_job_test.go
index 70936a93fa8..45ac8c529a3 100644
--- a/test/e2e/source_cron_job_test.go
+++ b/test/e2e/source_cron_job_test.go
@@ -21,7 +21,7 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/knative/eventing/test/base"
+	"github.com/knative/eventing/test/base/resources"
 	"github.com/knative/eventing/test/common"
 	"k8s.io/apimachinery/pkg/util/uuid"
 )
@@ -39,12 +39,12 @@ func TestCronJobSource(t *testing.T) {
 	defer TearDown(client)
 
 	// create event logger pod and service
-	loggerPod := base.EventLoggerPod(loggerPodName)
+	loggerPod := resources.EventLoggerPod(loggerPodName)
 	client.CreatePodOrFail(loggerPod, common.WithService(loggerPodName))
 
 	// create cron job source
 	data := fmt.Sprintf("TestCronJobSource %s", uuid.NewUUID())
-	sinkOption := base.WithSinkServiceForCronJobSource(loggerPodName)
+	sinkOption := resources.WithSinkServiceForCronJobSource(loggerPodName)
 	client.CreateCronJobSourceOrFail(cronJobSourceName, schedule, data, sinkOption)
 
 	// wait for all test resources to be ready
diff --git a/test/e2e/test_runner.go b/test/e2e/test_runner.go
index c9995ef3161..0d034d8dc21 100644
--- a/test/e2e/test_runner.go
+++ b/test/e2e/test_runner.go
@@ -119,7 +119,7 @@ func contains(features []common.Feature, feature common.Feature) bool {
 func getChannelTypeMeta(provisioner string, isCRD bool) *metav1.TypeMeta {
 	channelTypeMeta := common.ChannelTypeMeta
 	if isCRD {
-		channelTypeMeta = common.OperatorChannelMap[provisioner]
+		channelTypeMeta = common.ProvisionerChannelMap[provisioner]
 	}
 	return channelTypeMeta
 }
diff --git a/test/test_images/logevents/kodata/LICENSE b/test/test_images/logevents/kodata/LICENSE
deleted file mode 120000
index 14776154326..00000000000
--- a/test/test_images/logevents/kodata/LICENSE
+++ /dev/null
@@ -1 +0,0 @@
-../../../../LICENSE
\ No newline at end of file
diff --git a/test/test_images/logevents/kodata/LICENSE b/test/test_images/logevents/kodata/LICENSE
new file mode 100644
index 00000000000..14776154326
--- /dev/null
+++ b/test/test_images/logevents/kodata/LICENSE
@@ -0,0 +1 @@
+../../../../LICENSE
\ No newline at end of file
diff --git a/test/test_images/logevents/kodata/VENDOR-LICENSE b/test/test_images/logevents/kodata/VENDOR-LICENSE
deleted file mode 120000
index 7322c09d957..00000000000
--- a/test/test_images/logevents/kodata/VENDOR-LICENSE
+++ /dev/null
@@ -1 +0,0 @@
-../../../../third_party/VENDOR-LICENSE
\ No newline at end of file
diff --git a/test/test_images/logevents/kodata/VENDOR-LICENSE b/test/test_images/logevents/kodata/VENDOR-LICENSE
new file mode 100644
index 00000000000..7322c09d957
--- /dev/null
+++ b/test/test_images/logevents/kodata/VENDOR-LICENSE
@@ -0,0 +1 @@
+../../../../third_party/VENDOR-LICENSE
\ No newline at end of file