Repository: knative/client (go)

Issue:
Allow source creation with brokers
Allow `broker:default` and other brokers as sinks in source creation.

Possibly allow the default broker as a default sink for a source.



Pull request description:
Allow for http and broker sinks
<!--
Request Prow to automatically lint any go code in this PR:

/lint
-->
Refactors sink code to use a dynamic client to validate the sink, so we don't keep having to write the same parallel code. Paves the way for arbitrary user-configured sinks.

Fixes #570 

Also fixes most of #563 : missing arbitrary ns for sinks, but now has the sink as being in the same ns as the object being created.




Patch:
diff --git a/pkg/dynamic/client.go b/pkg/dynamic/client.go
index b71879d0..35d5ca52 100644
--- a/pkg/dynamic/client.go
+++ b/pkg/dynamic/client.go
@@ -42,6 +42,9 @@ type KnDynamicClient interface {
 
 	// ListSourceCRDs returns list of eventing sources CRDs
 	ListSourcesTypes() (*unstructured.UnstructuredList, error)
+
+	// RawClient returns the raw dynamic client interface
+	RawClient() dynamic.Interface
 }
 
 // knDynamicClient is a combination of client-go Dynamic client interface and namespace
@@ -87,3 +90,7 @@ func (c *knDynamicClient) ListSourcesTypes() (*unstructured.UnstructuredList, er
 	options.LabelSelector = sourcesLabels.String()
 	return c.ListCRDs(options)
 }
+
+func (c knDynamicClient) RawClient() dynamic.Interface {
+	return c.client
+}
diff --git a/pkg/dynamic/fake/fake.go b/pkg/dynamic/fake/fake.go
new file mode 100644
index 00000000..c8f21e5d
--- /dev/null
+++ b/pkg/dynamic/fake/fake.go
@@ -0,0 +1,34 @@
+// 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 fake
+
+import (
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8s_fake "k8s.io/client-go/dynamic/fake"
+
+	"knative.dev/client/pkg/dynamic"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
+)
+
+// CreateFakeKnDynamicClient gives you a dynamic client for testing contianing the given objects.
+func CreateFakeKnDynamicClient(testNamespace string, objects ...runtime.Object) dynamic.KnDynamicClient {
+	scheme := runtime.NewScheme()
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "serving.knative.dev", Version: "v1alpha1", Kind: "Service"}, &serving_v1alpha1.Service{})
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "eventing.knative.dev", Version: "v1alpha1", Kind: "Broker"}, &eventing_v1alpha1.Broker{})
+	client := k8s_fake.NewSimpleDynamicClient(scheme, objects...)
+	return dynamic.NewKnDynamicClient(client, testNamespace)
+}
diff --git a/pkg/kn/commands/flags/sink.go b/pkg/kn/commands/flags/sink.go
index 734895d0..5aed55f2 100644
--- a/pkg/kn/commands/flags/sink.go
+++ b/pkg/kn/commands/flags/sink.go
@@ -20,9 +20,11 @@ import (
 
 	"github.com/spf13/cobra"
 	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	kn_dynamic "knative.dev/client/pkg/dynamic"
+	"knative.dev/pkg/apis"
 	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
-
-	"knative.dev/client/pkg/serving/v1alpha1"
 )
 
 type SinkFlags struct {
@@ -33,26 +35,73 @@ func (i *SinkFlags) Add(cmd *cobra.Command) {
 	cmd.Flags().StringVarP(&i.sink, "sink", "s", "", "Addressable sink for events")
 }
 
-func (i *SinkFlags) ResolveSink(client v1alpha1.KnServingClient) (*duckv1beta1.Destination, error) {
+// SinkPrefixes maps prefixes used for sinks to their GroupVersionResources.
+var SinkPrefixes = map[string]schema.GroupVersionResource{
+	"broker": {
+		Resource: "brokers",
+		Group:    "eventing.knative.dev",
+		Version:  "v1alpha1",
+	},
+	"service": {
+		Resource: "services",
+		Group:    "serving.knative.dev",
+		Version:  "v1alpha1",
+	},
+	// Shorthand alias for service
+	"svc": {
+		Resource: "services",
+		Group:    "serving.knative.dev",
+		Version:  "v1alpha1",
+	},
+}
+
+// ResolveSink returns the Destination referred to by the flags in the acceptor.
+// It validates that any object the user is referring to exists.
+func (i *SinkFlags) ResolveSink(knclient kn_dynamic.KnDynamicClient, namespace string) (*duckv1beta1.Destination, error) {
+	client := knclient.RawClient()
 	if i.sink == "" {
 		return nil, nil
 	}
 
-	if strings.HasPrefix(i.sink, "svc:") {
-		serviceName := i.sink[4:]
-		service, err := client.GetService(serviceName)
+	prefix, name := parseSink(i.sink)
+	if prefix == "" {
+		// URI target
+		uri, err := apis.ParseURL(name)
 		if err != nil {
 			return nil, err
 		}
-		return &duckv1beta1.Destination{
-			Ref: &v1.ObjectReference{
-				Kind:       service.Kind,
-				APIVersion: service.APIVersion,
-				Name:       service.Name,
-				Namespace:  service.Namespace,
-			},
-		}, nil
+		return &duckv1beta1.Destination{URI: uri}, nil
+	}
+	typ, ok := SinkPrefixes[prefix]
+	if !ok {
+		return nil, fmt.Errorf("Not supported sink type: %s", i.sink)
+	}
+	obj, err := client.Resource(typ).Namespace(namespace).Get(name, metav1.GetOptions{})
+	if err != nil {
+		return nil, err
 	}
 
-	return nil, fmt.Errorf("Not supported sink type: %s", i.sink)
+	return &duckv1beta1.Destination{
+		Ref: &v1.ObjectReference{
+			Kind:       obj.GetKind(),
+			APIVersion: obj.GetAPIVersion(),
+			Name:       obj.GetName(),
+			Namespace:  namespace,
+		},
+	}, nil
+
+}
+
+// parseSink takes the string given by the user into the prefix and the name of
+// the object. If the user put a URI instead, the prefix is empty and the name
+// is the whole URI.
+func parseSink(sink string) (string, string) {
+	parts := strings.SplitN(sink, ":", 2)
+	if len(parts) == 1 {
+		return "svc", parts[0]
+	} else if parts[0] == "http" || parts[0] == "https" {
+		return "", sink
+	} else {
+		return parts[0], parts[1]
+	}
 }
diff --git a/pkg/kn/commands/source/apiserver/create.go b/pkg/kn/commands/source/apiserver/create.go
index 797fd105..6425dd1e 100644
--- a/pkg/kn/commands/source/apiserver/create.go
+++ b/pkg/kn/commands/source/apiserver/create.go
@@ -51,13 +51,11 @@ func NewAPIServerCreateCommand(p *commands.KnParams) *cobra.Command {
 
 			namespace := apiSourceClient.Namespace()
 
-			// create Serving client for resolving service sink
-			servingClient, err := p.NewServingClient(namespace)
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
-
-			objectRef, err := sinkFlags.ResolveSink(servingClient)
+			objectRef, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 			if err != nil {
 				return fmt.Errorf(
 					"cannot create ApiServerSource '%s' in namespace '%s' "+
diff --git a/pkg/kn/commands/source/apiserver/update.go b/pkg/kn/commands/source/apiserver/update.go
index e51ba091..5491192c 100644
--- a/pkg/kn/commands/source/apiserver/update.go
+++ b/pkg/kn/commands/source/apiserver/update.go
@@ -49,7 +49,7 @@ func NewAPIServerUpdateCommand(p *commands.KnParams) *cobra.Command {
 				return err
 			}
 
-			servingClient, err := p.NewServingClient(namespace)
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
@@ -82,7 +82,7 @@ func NewAPIServerUpdateCommand(p *commands.KnParams) *cobra.Command {
 			}
 
 			if cmd.Flags().Changed("sink") {
-				objectRef, err := sinkFlags.ResolveSink(servingClient)
+				objectRef, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 				if err != nil {
 					return err
 				}
diff --git a/pkg/kn/commands/source/cronjob/create.go b/pkg/kn/commands/source/cronjob/create.go
index 84c7e1be..cee867f9 100644
--- a/pkg/kn/commands/source/cronjob/create.go
+++ b/pkg/kn/commands/source/cronjob/create.go
@@ -49,12 +49,16 @@ func NewCronJobCreateCommand(p *commands.KnParams) *cobra.Command {
 				return err
 			}
 
-			servingClient, err := p.NewServingClient(cronSourceClient.Namespace())
+			namespace, err := p.GetNamespace(cmd)
+			if err != nil {
+				return err
+			}
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
 
-			destination, err := sinkFlags.ResolveSink(servingClient)
+			destination, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 			if err != nil {
 				return err
 			}
diff --git a/pkg/kn/commands/source/cronjob/update.go b/pkg/kn/commands/source/cronjob/update.go
index 3741a3b5..8e594429 100644
--- a/pkg/kn/commands/source/cronjob/update.go
+++ b/pkg/kn/commands/source/cronjob/update.go
@@ -48,7 +48,11 @@ func NewCronJobUpdateCommand(p *commands.KnParams) *cobra.Command {
 				return err
 			}
 
-			servingClient, err := p.NewServingClient(cronSourceClient.Namespace())
+			namespace, err := p.GetNamespace(cmd)
+			if err != nil {
+				return err
+			}
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
@@ -66,7 +70,7 @@ func NewCronJobUpdateCommand(p *commands.KnParams) *cobra.Command {
 				b.Data(cronUpdateFlags.data)
 			}
 			if cmd.Flags().Changed("sink") {
-				destination, err := sinkFlags.ResolveSink(servingClient)
+				destination, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 				if err != nil {
 					return err
 				}
diff --git a/pkg/kn/commands/trigger/create.go b/pkg/kn/commands/trigger/create.go
index 93f7d1a0..5d82b122 100644
--- a/pkg/kn/commands/trigger/create.go
+++ b/pkg/kn/commands/trigger/create.go
@@ -50,17 +50,17 @@ func NewTriggerCreateCommand(p *commands.KnParams) *cobra.Command {
 				return err
 			}
 
-			eventingClient, err := p.NewEventingClient(namespace)
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
 
-			servingClient, err := p.NewServingClient(namespace)
+			eventingClient, err := p.NewEventingClient(namespace)
 			if err != nil {
 				return err
 			}
 
-			objectRef, err := sinkFlags.ResolveSink(servingClient)
+			objectRef, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 			if err != nil {
 				return fmt.Errorf(
 					"cannot create trigger '%s' in namespace '%s' "+
diff --git a/pkg/kn/commands/trigger/update.go b/pkg/kn/commands/trigger/update.go
index 5545197d..be63a63d 100644
--- a/pkg/kn/commands/trigger/update.go
+++ b/pkg/kn/commands/trigger/update.go
@@ -60,8 +60,7 @@ func NewTriggerUpdateCommand(p *commands.KnParams) *cobra.Command {
 			if err != nil {
 				return err
 			}
-
-			servingClient, err := p.NewServingClient(namespace)
+			dynamicClient, err := p.NewDynamicClient(namespace)
 			if err != nil {
 				return err
 			}
@@ -91,7 +90,7 @@ func NewTriggerUpdateCommand(p *commands.KnParams) *cobra.Command {
 				}
 			}
 			if cmd.Flags().Changed("sink") {
-				destination, err := sinkFlags.ResolveSink(servingClient)
+				destination, err := sinkFlags.ResolveSink(dynamicClient, namespace)
 				if err != nil {
 					return err
 				}


Test patch:
diff --git a/pkg/dynamic/client_test.go b/pkg/dynamic/client_test.go
index f09b8f95..96c4214b 100644
--- a/pkg/dynamic/client_test.go
+++ b/pkg/dynamic/client_test.go
@@ -22,7 +22,10 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/client-go/dynamic/fake"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	k8s_fake "k8s.io/client-go/dynamic/fake"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 )
 
 const testNamespace = "testns"
@@ -43,18 +46,14 @@ func newUnstructured(name string) *unstructured.Unstructured {
 	}
 }
 
-func createFakeKnDynamicClient(objects ...runtime.Object) KnDynamicClient {
-	client := fake.NewSimpleDynamicClient(runtime.NewScheme(), objects...)
-	return NewKnDynamicClient(client, testNamespace)
-}
-
 func TestNamespace(t *testing.T) {
-	client := createFakeKnDynamicClient(newUnstructured("foo"))
+	client := createFakeKnDynamicClient(testNamespace, newUnstructured("foo"))
 	assert.Equal(t, client.Namespace(), testNamespace)
 }
 
 func TestListCRDs(t *testing.T) {
 	client := createFakeKnDynamicClient(
+		testNamespace,
 		newUnstructured("foo"),
 		newUnstructured("bar"),
 	)
@@ -84,6 +83,7 @@ func TestListCRDs(t *testing.T) {
 
 func TestListSourceTypes(t *testing.T) {
 	client := createFakeKnDynamicClient(
+		testNamespace,
 		newUnstructured("foo"),
 		newUnstructured("bar"),
 	)
@@ -99,3 +99,13 @@ func TestListSourceTypes(t *testing.T) {
 		assert.Equal(t, uList.Items[1].GetName(), "bar")
 	})
 }
+
+// createFakeKnDynamicClient gives you a dynamic client for testing contianing the given objects.
+// See also the one in the fake package. Duplicated here to avoid a dependency loop.
+func createFakeKnDynamicClient(testNamespace string, objects ...runtime.Object) KnDynamicClient {
+	scheme := runtime.NewScheme()
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "serving.knative.dev", Version: "v1alpha1", Kind: "Service"}, &serving_v1alpha1.Service{})
+	scheme.AddKnownTypeWithName(schema.GroupVersionKind{Group: "eventing.knative.dev", Version: "v1alpha1", Kind: "Broker"}, &eventing_v1alpha1.Broker{})
+	client := k8s_fake.NewSimpleDynamicClient(scheme, objects...)
+	return NewKnDynamicClient(client, testNamespace)
+}
diff --git a/pkg/kn/commands/flags/sink_test.go b/pkg/kn/commands/flags/sink_test.go
new file mode 100644
index 00000000..35ea0ede
--- /dev/null
+++ b/pkg/kn/commands/flags/sink_test.go
@@ -0,0 +1,81 @@
+// 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 flags
+
+import (
+	"testing"
+
+	"gotest.tools/assert"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
+	eventing_v1alpha1 "knative.dev/eventing/pkg/apis/eventing/v1alpha1"
+	"knative.dev/pkg/apis"
+	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
+	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
+)
+
+type resolveCase struct {
+	sink        string
+	destination *duckv1beta1.Destination
+	errContents string
+}
+
+func TestResolve(t *testing.T) {
+	targetExampleCom, err := apis.ParseURL("http://target.example.com")
+	mysvc := &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	}
+	defaultBroker := &eventing_v1alpha1.Broker{
+		TypeMeta:   metav1.TypeMeta{Kind: "Broker", APIVersion: "eventing.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "default", Namespace: "default"},
+	}
+
+	assert.NilError(t, err)
+	cases := []resolveCase{
+		{"svc:mysvc", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Service",
+				APIVersion: "serving.knative.dev/v1alpha1",
+				Name:       "mysvc",
+				Namespace:  "default"}}, ""},
+		{"service:mysvc", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Service",
+				APIVersion: "serving.knative.dev/v1alpha1",
+				Name:       "mysvc",
+				Namespace:  "default"}}, ""},
+		{"svc:absent", nil, "\"absent\" not found"},
+		{"broker:default", &duckv1beta1.Destination{
+			Ref: &v1.ObjectReference{Kind: "Broker",
+				APIVersion: "eventing.knative.dev/v1alpha1",
+				Name:       "default",
+				Namespace:  "default",
+			}}, ""},
+		{"http://target.example.com", &duckv1beta1.Destination{
+			URI: targetExampleCom,
+		}, ""},
+	}
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", mysvc, defaultBroker)
+	for _, c := range cases {
+		i := &SinkFlags{c.sink}
+		result, err := i.ResolveSink(dynamicClient, "default")
+		if c.destination != nil {
+			assert.DeepEqual(t, result, c.destination)
+			assert.NilError(t, err)
+		} else {
+			assert.ErrorContains(t, err, c.errContents)
+		}
+	}
+}
diff --git a/pkg/kn/commands/source/apiserver/apiserver_test.go b/pkg/kn/commands/source/apiserver/apiserver_test.go
index 09777f24..cc210ccf 100644
--- a/pkg/kn/commands/source/apiserver/apiserver_test.go
+++ b/pkg/kn/commands/source/apiserver/apiserver_test.go
@@ -19,12 +19,12 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/clientcmd"
+	kn_dynamic "knative.dev/client/pkg/dynamic"
 	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
 	duckv1beta1 "knative.dev/pkg/apis/duck/v1beta1"
 
 	knsource_v1alpha1 "knative.dev/client/pkg/eventing/sources/v1alpha1"
 	"knative.dev/client/pkg/kn/commands"
-	knserving_v1alpha1 "knative.dev/client/pkg/serving/v1alpha1"
 )
 
 const testNamespace = "default"
@@ -55,14 +55,14 @@ current-context: x
 	}
 }
 
-func executeAPIServerSourceCommand(apiServerSourceClient knsource_v1alpha1.KnAPIServerSourcesClient, servingClient knserving_v1alpha1.KnServingClient, args ...string) (string, error) {
+func executeAPIServerSourceCommand(apiServerSourceClient knsource_v1alpha1.KnAPIServerSourcesClient, dynamicClient kn_dynamic.KnDynamicClient, args ...string) (string, error) {
 	knParams := &commands.KnParams{}
 	knParams.ClientConfig = blankConfig
 
 	output := new(bytes.Buffer)
 	knParams.Output = output
-	knParams.NewServingClient = func(namespace string) (knserving_v1alpha1.KnServingClient, error) {
-		return servingClient, nil
+	knParams.NewDynamicClient = func(namespace string) (kn_dynamic.KnDynamicClient, error) {
+		return dynamicClient, nil
 	}
 
 	cmd := NewAPIServerCommand(knParams)
@@ -92,8 +92,10 @@ func createAPIServerSource(name, resourceKind, resourceVersion, serviceAccount,
 
 	sink := &duckv1beta1.Destination{
 		Ref: &corev1.ObjectReference{
-			Kind: "Service",
-			Name: service,
+			Kind:       "Service",
+			Name:       service,
+			APIVersion: "serving.knative.dev/v1alpha1",
+			Namespace:  "default",
 		}}
 
 	return knsource_v1alpha1.NewAPIServerSourceBuilder(name).
diff --git a/pkg/kn/commands/source/apiserver/create_test.go b/pkg/kn/commands/source/apiserver/create_test.go
index 176e17c1..880ae7c4 100644
--- a/pkg/kn/commands/source/apiserver/create_test.go
+++ b/pkg/kn/commands/source/apiserver/create_test.go
@@ -15,51 +15,41 @@
 package apiserver
 
 import (
-	"errors"
 	"testing"
 
 	"gotest.tools/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
 	knsources_v1alpha1 "knative.dev/client/pkg/eventing/sources/v1alpha1"
-	knserving_client "knative.dev/client/pkg/serving/v1alpha1"
 	"knative.dev/client/pkg/util"
 	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 )
 
 func TestCreateApiServerSource(t *testing.T) {
-
+	testsvc := &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "testsvc", Namespace: "default"},
+	}
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", testsvc)
 	apiServerClient := knsources_v1alpha1.NewMockKnAPIServerSourceClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
-
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("testsvc", &serving_v1alpha1.Service{
-		TypeMeta:   metav1.TypeMeta{Kind: "Service"},
-		ObjectMeta: metav1.ObjectMeta{Name: "testsvc"},
-	}, nil)
 
 	apiServerRecorder := apiServerClient.Recorder()
 	apiServerRecorder.CreateAPIServerSource(createAPIServerSource("testsource", "Event", "v1", "testsa", "Ref", "testsvc", false), nil)
 
-	out, err := executeAPIServerSourceCommand(apiServerClient, servingClient, "create", "testsource", "--resource", "Event:v1:false", "--service-account", "testsa", "--sink", "svc:testsvc", "--mode", "Ref")
+	out, err := executeAPIServerSourceCommand(apiServerClient, dynamicClient, "create", "testsource", "--resource", "Event:v1:false", "--service-account", "testsa", "--sink", "svc:testsvc", "--mode", "Ref")
 	assert.NilError(t, err, "ApiServer source should be created")
 	util.ContainsAll(out, "created", "default", "testsource")
 
 	apiServerRecorder.Validate()
-	servingRecorder.Validate()
 }
 
 func TestSinkNotFoundError(t *testing.T) {
-	servingClient := knserving_client.NewMockKnServiceClient(t)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default")
 	apiServerClient := knsources_v1alpha1.NewMockKnAPIServerSourceClient(t)
-
-	errorMsg := "cannot create ApiServerSource 'testsource' in namespace 'default' because: no Service svc found"
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("testsvc", nil, errors.New("no Service svc found"))
-
-	out, err := executeAPIServerSourceCommand(apiServerClient, servingClient, "create", "testsource", "--resource", "Event:v1:false", "--service-account", "testsa", "--sink", "svc:testsvc", "--mode", "Ref")
+	errorMsg := "cannot create ApiServerSource 'testsource' in namespace 'default' because: services.serving.knative.dev \"testsvc\" not found"
+	out, err := executeAPIServerSourceCommand(apiServerClient, dynamicClient, "create", "testsource", "--resource", "Event:v1:false", "--service-account", "testsa", "--sink", "svc:testsvc", "--mode", "Ref")
 	assert.Error(t, err, errorMsg)
 	assert.Assert(t, util.ContainsAll(out, errorMsg, "Usage"))
-	servingRecorder.Validate()
 }
 
 func TestNoSinkError(t *testing.T) {
diff --git a/pkg/kn/commands/source/apiserver/update_test.go b/pkg/kn/commands/source/apiserver/update_test.go
index a8df007a..7c03dda0 100644
--- a/pkg/kn/commands/source/apiserver/update_test.go
+++ b/pkg/kn/commands/source/apiserver/update_test.go
@@ -22,33 +22,29 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
 	knsources_v1alpha1 "knative.dev/client/pkg/eventing/sources/v1alpha1"
-	knserving_client "knative.dev/client/pkg/serving/v1alpha1"
 	"knative.dev/client/pkg/util"
 )
 
 func TestApiServerSourceUpdate(t *testing.T) {
 	apiServerClient := knsources_v1alpha1.NewMockKnAPIServerSourceClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "svc2", Namespace: "default"},
+	})
 
 	apiServerRecorder := apiServerClient.Recorder()
-	servingRecorder := servingClient.Recorder()
 
 	present := createAPIServerSource("testsource", "Event", "v1", "testsa1", "Ref", "svc1", false)
 	apiServerRecorder.GetAPIServerSource("testsource", present, nil)
 
-	servingRecorder.GetService("svc2", &serving_v1alpha1.Service{
-		TypeMeta:   metav1.TypeMeta{Kind: "Service"},
-		ObjectMeta: metav1.ObjectMeta{Name: "svc2"},
-	}, nil)
-
 	updated := createAPIServerSource("testsource", "Event", "v1", "testsa2", "Ref", "svc2", false)
 	apiServerRecorder.UpdateAPIServerSource(updated, nil)
 
-	output, err := executeAPIServerSourceCommand(apiServerClient, servingClient, "update", "testsource", "--service-account", "testsa2", "--sink", "svc:svc2")
+	output, err := executeAPIServerSourceCommand(apiServerClient, dynamicClient, "update", "testsource", "--service-account", "testsa2", "--sink", "svc:svc2")
 	assert.NilError(t, err)
 	assert.Assert(t, util.ContainsAll(output, "testsource", "updated", "default"))
 
 	apiServerRecorder.Validate()
-	servingRecorder.Validate()
 }
diff --git a/pkg/kn/commands/source/cronjob/create_test.go b/pkg/kn/commands/source/cronjob/create_test.go
index badb05c8..2696f98d 100644
--- a/pkg/kn/commands/source/cronjob/create_test.go
+++ b/pkg/kn/commands/source/cronjob/create_test.go
@@ -15,52 +15,44 @@
 package cronjob
 
 import (
-	"errors"
 	"testing"
 
 	"gotest.tools/assert"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
 	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 
 	v1alpha12 "knative.dev/client/pkg/eventing/sources/v1alpha1"
-	knservingclient "knative.dev/client/pkg/serving/v1alpha1"
 	"knative.dev/client/pkg/util"
 )
 
 func TestSimpleCreateCronJobSource(t *testing.T) {
+	mysvc := &serving_v1alpha1.Service{
+		TypeMeta:   v1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: v1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	}
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", mysvc)
 
-	servingClient := knservingclient.NewMockKnServiceClient(t)
 	cronjobClient := v1alpha12.NewMockKnCronJobSourceClient(t)
 
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", &serving_v1alpha1.Service{
-		TypeMeta:   v1.TypeMeta{Kind: "Service"},
-		ObjectMeta: v1.ObjectMeta{Name: "mysvc"},
-	}, nil)
-
 	cronJobRecorder := cronjobClient.Recorder()
 	cronJobRecorder.CreateCronJobSource(createCronJobSource("testsource", "* * * * */2", "maxwell", "mysvc"), nil)
 
-	out, err := executeCronJobSourceCommand(cronjobClient, servingClient, "create", "--sink", "svc:mysvc", "--schedule", "* * * * */2", "--data", "maxwell", "testsource")
+	out, err := executeCronJobSourceCommand(cronjobClient, dynamicClient, "create", "--sink", "svc:mysvc", "--schedule", "* * * * */2", "--data", "maxwell", "testsource")
 	assert.NilError(t, err, "Source should have been created")
 	util.ContainsAll(out, "created", "default", "testsource")
 
 	cronJobRecorder.Validate()
-	servingRecorder.Validate()
 }
 
 func TestNoSinkError(t *testing.T) {
-	servingClient := knservingclient.NewMockKnServiceClient(t)
 	cronjobClient := v1alpha12.NewMockKnCronJobSourceClient(t)
 
-	errorMsg := "no Service mysvc found"
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", nil, errors.New(errorMsg))
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default")
 
-	out, err := executeCronJobSourceCommand(cronjobClient, servingClient, "create", "--sink", "svc:mysvc", "--schedule", "* * * * */2", "--data", "maxwell", "testsource")
-	assert.Error(t, err, errorMsg)
-	assert.Assert(t, util.ContainsAll(out, errorMsg, "Usage"))
-	servingRecorder.Validate()
+	out, err := executeCronJobSourceCommand(cronjobClient, dynamicClient, "create", "--sink", "svc:mysvc", "--schedule", "* * * * */2", "--data", "maxwell", "testsource")
+	assert.Error(t, err, "services.serving.knative.dev \"mysvc\" not found")
+	assert.Assert(t, util.ContainsAll(out, "Usage"))
 }
 
 func TestNoSinkGivenError(t *testing.T) {
diff --git a/pkg/kn/commands/source/cronjob/cronjob_test.go b/pkg/kn/commands/source/cronjob/cronjob_test.go
index e5afa17d..8a61cf10 100644
--- a/pkg/kn/commands/source/cronjob/cronjob_test.go
+++ b/pkg/kn/commands/source/cronjob/cronjob_test.go
@@ -22,9 +22,9 @@ import (
 	"knative.dev/eventing/pkg/apis/sources/v1alpha1"
 	"knative.dev/pkg/apis/duck/v1beta1"
 
+	kn_dynamic "knative.dev/client/pkg/dynamic"
 	source_client_v1alpha1 "knative.dev/client/pkg/eventing/sources/v1alpha1"
 	"knative.dev/client/pkg/kn/commands"
-	serving_client_v1alpha1 "knative.dev/client/pkg/serving/v1alpha1"
 )
 
 // Helper methods
@@ -54,14 +54,14 @@ current-context: x
 	}
 }
 
-func executeCronJobSourceCommand(cronJobSourceClient source_client_v1alpha1.KnCronJobSourcesClient, servingClient serving_client_v1alpha1.KnServingClient, args ...string) (string, error) {
+func executeCronJobSourceCommand(cronJobSourceClient source_client_v1alpha1.KnCronJobSourcesClient, dynamicClient kn_dynamic.KnDynamicClient, args ...string) (string, error) {
 	knParams := &commands.KnParams{}
 	knParams.ClientConfig = blankConfig
 
 	output := new(bytes.Buffer)
 	knParams.Output = output
-	knParams.NewServingClient = func(namespace string) (serving_client_v1alpha1.KnServingClient, error) {
-		return servingClient, nil
+	knParams.NewDynamicClient = func(namespace string) (kn_dynamic.KnDynamicClient, error) {
+		return dynamicClient, nil
 	}
 
 	cmd := NewCronJobCommand(knParams)
@@ -84,7 +84,7 @@ func cleanupCronJobMockClient() {
 
 func createCronJobSource(name, schedule, data, service string) *v1alpha1.CronJobSource {
 	sink := &v1beta1.Destination{
-		Ref: &corev1.ObjectReference{Name: service, Kind: "Service"},
+		Ref: &corev1.ObjectReference{Name: service, Kind: "Service", Namespace: "default", APIVersion: "serving.knative.dev/v1alpha1"},
 	}
 	return source_client_v1alpha1.NewCronJobSourceBuilder(name).Schedule(schedule).Data(data).Sink(sink).Build()
 }
diff --git a/pkg/kn/commands/source/cronjob/update_test.go b/pkg/kn/commands/source/cronjob/update_test.go
index b01821cd..aeee77ca 100644
--- a/pkg/kn/commands/source/cronjob/update_test.go
+++ b/pkg/kn/commands/source/cronjob/update_test.go
@@ -43,8 +43,10 @@ func TestSimpleUpdate(t *testing.T) {
 			Data:     "maxwell",
 			Sink: &v1beta1.Destination{
 				Ref: &corev1.ObjectReference{
-					Kind: "Service",
-					Name: "mysvc",
+					Kind:       "Service",
+					Name:       "mysvc",
+					Namespace:  "default",
+					APIVersion: "serving.knative.dev/v1alpha1",
 				},
 			},
 		},
diff --git a/pkg/kn/commands/trigger/create_test.go b/pkg/kn/commands/trigger/create_test.go
index b8d9a4ff..e6d1ebac 100644
--- a/pkg/kn/commands/trigger/create_test.go
+++ b/pkg/kn/commands/trigger/create_test.go
@@ -15,14 +15,13 @@
 package trigger
 
 import (
-	"errors"
 	"fmt"
 	"testing"
 
 	"gotest.tools/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
 	eventing_client "knative.dev/client/pkg/eventing/v1alpha1"
-	knserving_client "knative.dev/client/pkg/serving/v1alpha1"
 	"knative.dev/client/pkg/util"
 	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 )
@@ -33,39 +32,32 @@ var (
 
 func TestTriggerCreate(t *testing.T) {
 	eventingClient := eventing_client.NewMockKnEventingClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
-
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", &serving_v1alpha1.Service{
-		TypeMeta:   metav1.TypeMeta{Kind: "Service"},
-		ObjectMeta: metav1.ObjectMeta{Name: "mysvc"},
-	}, nil)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	})
 
 	eventingRecorder := eventingClient.Recorder()
 	eventingRecorder.CreateTrigger(createTrigger("default", triggerName, map[string]string{"type": "dev.knative.foo"}, "mybroker", "mysvc"), nil)
 
-	out, err := executeTriggerCommand(eventingClient, servingClient, "create", triggerName, "--broker", "mybroker",
+	out, err := executeTriggerCommand(eventingClient, dynamicClient, "create", triggerName, "--broker", "mybroker",
 		"--filter", "type=dev.knative.foo", "--sink", "svc:mysvc")
 	assert.NilError(t, err, "Trigger should be created")
 	util.ContainsAll(out, "Trigger", triggerName, "created", "namespace", "default")
 
 	eventingRecorder.Validate()
-	servingRecorder.Validate()
 }
 
 func TestSinkNotFoundError(t *testing.T) {
 	eventingClient := eventing_client.NewMockKnEventingClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default")
 
-	errorMsg := fmt.Sprintf("cannot create trigger '%s' in namespace 'default' because: no Service mysvc found", triggerName)
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", nil, errors.New("no Service mysvc found"))
+	errorMsg := fmt.Sprintf("cannot create trigger '%s' in namespace 'default' because: services.serving.knative.dev \"mysvc\" not found", triggerName)
 
-	out, err := executeTriggerCommand(eventingClient, servingClient, "create", triggerName, "--broker", "mybroker",
+	out, err := executeTriggerCommand(eventingClient, dynamicClient, "create", triggerName, "--broker", "mybroker",
 		"--filter", "type=dev.knative.foo", "--sink", "svc:mysvc")
 	assert.Error(t, err, errorMsg)
 	assert.Assert(t, util.ContainsAll(out, errorMsg, "Usage"))
-	servingRecorder.Validate()
 }
 
 func TestNoSinkError(t *testing.T) {
@@ -84,22 +76,18 @@ func TestNoFilterError(t *testing.T) {
 
 func TestTriggerCreateMultipleFilter(t *testing.T) {
 	eventingClient := eventing_client.NewMockKnEventingClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
-
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", &serving_v1alpha1.Service{
-		TypeMeta:   metav1.TypeMeta{Kind: "Service"},
-		ObjectMeta: metav1.ObjectMeta{Name: "mysvc"},
-	}, nil)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	})
 
 	eventingRecorder := eventingClient.Recorder()
 	eventingRecorder.CreateTrigger(createTrigger("default", triggerName, map[string]string{"type": "dev.knative.foo", "source": "event.host"}, "mybroker", "mysvc"), nil)
 
-	out, err := executeTriggerCommand(eventingClient, servingClient, "create", triggerName, "--broker", "mybroker",
+	out, err := executeTriggerCommand(eventingClient, dynamicClient, "create", triggerName, "--broker", "mybroker",
 		"--filter", "type=dev.knative.foo", "--filter", "source=event.host", "--sink", "svc:mysvc")
 	assert.NilError(t, err, "Trigger should be created")
 	util.ContainsAll(out, "Trigger", triggerName, "created", "namespace", "default")
 
 	eventingRecorder.Validate()
-	servingRecorder.Validate()
 }
diff --git a/pkg/kn/commands/trigger/list_test.go b/pkg/kn/commands/trigger/list_test.go
index 1b355d6b..f89c4e7b 100644
--- a/pkg/kn/commands/trigger/list_test.go
+++ b/pkg/kn/commands/trigger/list_test.go
@@ -44,7 +44,7 @@ func TestTriggerList(t *testing.T) {
 	triggerList := &v1alpha1.TriggerList{Items: []v1alpha1.Trigger{*trigger1, *trigger2, *trigger3}}
 	eventingRecorder.ListTriggers(triggerList, nil)
 
-	output, err := executeTriggerCommand(eventingClient, servingClient, "list")
+	output, err := executeTriggerCommand(eventingClient, nil, "list")
 	assert.NilError(t, err)
 
 	outputLines := strings.Split(output, "\n")
@@ -85,7 +85,7 @@ func TestTriggerListAllNamespace(t *testing.T) {
 	triggerList := &v1alpha1.TriggerList{Items: []v1alpha1.Trigger{*trigger1, *trigger2, *trigger3}}
 	eventingRecorder.ListTriggers(triggerList, nil)
 
-	output, err := executeTriggerCommand(eventingClient, servingClient, "list", "--all-namespaces")
+	output, err := executeTriggerCommand(eventingClient, nil, "list", "--all-namespaces")
 	assert.NilError(t, err)
 
 	outputLines := strings.Split(output, "\n")
diff --git a/pkg/kn/commands/trigger/trigger_test.go b/pkg/kn/commands/trigger/trigger_test.go
index 6cff17b7..9c09d00d 100644
--- a/pkg/kn/commands/trigger/trigger_test.go
+++ b/pkg/kn/commands/trigger/trigger_test.go
@@ -19,14 +19,12 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/client-go/tools/clientcmd"
-
+	kn_dynamic "knative.dev/client/pkg/dynamic"
+	eventc_v1alpha1 "knative.dev/client/pkg/eventing/v1alpha1"
+	"knative.dev/client/pkg/kn/commands"
 	"knative.dev/eventing/pkg/apis/eventing/v1alpha1"
 	"knative.dev/pkg/apis"
 	duckv1 "knative.dev/pkg/apis/duck/v1"
-
-	eventc_v1alpha1 "knative.dev/client/pkg/eventing/v1alpha1"
-	"knative.dev/client/pkg/kn/commands"
-	serving_client_v1alpha1 "knative.dev/client/pkg/serving/v1alpha1"
 )
 
 // Helper methods
@@ -54,15 +52,16 @@ current-context: x
 	}
 }
 
-func executeTriggerCommand(triggerClient eventc_v1alpha1.KnEventingClient, servingClient serving_client_v1alpha1.KnServingClient, args ...string) (string, error) {
+func executeTriggerCommand(triggerClient eventc_v1alpha1.KnEventingClient, dynamicClient kn_dynamic.KnDynamicClient, args ...string) (string, error) {
 	knParams := &commands.KnParams{}
 	knParams.ClientConfig = blankConfig
 
 	output := new(bytes.Buffer)
 	knParams.Output = output
-	knParams.NewServingClient = func(namespace string) (serving_client_v1alpha1.KnServingClient, error) {
-		return servingClient, nil
+	knParams.NewDynamicClient = func(namespace string) (kn_dynamic.KnDynamicClient, error) {
+		return dynamicClient, nil
 	}
+
 	knParams.NewEventingClient = func(namespace string) (eventc_v1alpha1.KnEventingClient, error) {
 		return triggerClient, nil
 	}
@@ -87,8 +86,10 @@ func createTrigger(namespace string, name string, filters map[string]string, bro
 
 	triggerBuilder.Subscriber(&duckv1.Destination{
 		Ref: &corev1.ObjectReference{
-			Name: svcname,
-			Kind: "Service",
+			Name:       svcname,
+			Kind:       "Service",
+			Namespace:  "default",
+			APIVersion: "serving.knative.dev/v1alpha1",
 		},
 	})
 	return triggerBuilder.Build()
diff --git a/pkg/kn/commands/trigger/update_test.go b/pkg/kn/commands/trigger/update_test.go
index 0fba1db8..31b699df 100644
--- a/pkg/kn/commands/trigger/update_test.go
+++ b/pkg/kn/commands/trigger/update_test.go
@@ -20,21 +20,18 @@ import (
 
 	"gotest.tools/assert"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	dynamic_fake "knative.dev/client/pkg/dynamic/fake"
 	eventing_client "knative.dev/client/pkg/eventing/v1alpha1"
-	knserving_client "knative.dev/client/pkg/serving/v1alpha1"
 	"knative.dev/client/pkg/util"
 	serving_v1alpha1 "knative.dev/serving/pkg/apis/serving/v1alpha1"
 )
 
 func TestTriggerUpdate(t *testing.T) {
 	eventingClient := eventing_client.NewMockKnEventingClient(t)
-	servingClient := knserving_client.NewMockKnServiceClient(t)
-
-	servingRecorder := servingClient.Recorder()
-	servingRecorder.GetService("mysvc", &serving_v1alpha1.Service{
-		TypeMeta:   metav1.TypeMeta{Kind: "Service"},
-		ObjectMeta: metav1.ObjectMeta{Name: "mysvc"},
-	}, nil)
+	dynamicClient := dynamic_fake.CreateFakeKnDynamicClient("default", &serving_v1alpha1.Service{
+		TypeMeta:   metav1.TypeMeta{Kind: "Service", APIVersion: "serving.knative.dev/v1alpha1"},
+		ObjectMeta: metav1.ObjectMeta{Name: "mysvc", Namespace: "default"},
+	})
 
 	eventingRecorder := eventingClient.Recorder()
 	present := createTrigger("default", triggerName, map[string]string{"type": "dev.knative.foo"}, "mybroker", "mysvc")
@@ -42,13 +39,12 @@ func TestTriggerUpdate(t *testing.T) {
 	eventingRecorder.GetTrigger(triggerName, present, nil)
 	eventingRecorder.UpdateTrigger(updated, nil)
 
-	out, err := executeTriggerCommand(eventingClient, servingClient, "update", triggerName,
+	out, err := executeTriggerCommand(eventingClient, dynamicClient, "update", triggerName,
 		"--filter", "type=dev.knative.new", "--sink", "svc:mysvc")
 	assert.NilError(t, err, "Trigger should be updated")
 	util.ContainsAll(out, "Trigger", triggerName, "updated", "namespace", "default")
 
 	eventingRecorder.Validate()
-	servingRecorder.Validate()
 }
 
 func TestTriggerUpdateWithError(t *testing.T) {