Repository: knative-sandbox/eventing-kafka-broker
Pull request #2826 (2022-12-05 15:31:29)

[release-1.8] Fix namespaced broker unit tests
This is an automated cherry-pick of #2824

```release-note

```

knative-prow[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2826#" title="Author self-approved">knative-prow-robot</a>*
**Once this PR has been reviewed and has the lgtm label**, please assign evankanderson for approval by writing `/assign @evankanderson` in a comment. For more information see [the Kubernetes Code Review Process](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process).

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.8/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["evankanderson"]} -->
codecov[bot]: # [Codecov](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) Report
> Merging [#2826](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (b99ad68) into [release-1.8](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/commit/df5fead5c7e3fee528fe4b0376068294d0831935?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) (df5fead) will **increase** coverage by `22.94%`.
> The diff coverage is `n/a`.

```diff
@@                Coverage Diff                 @@
##             release-1.8    #2826       +/-   ##
==================================================
+ Coverage          58.04%   80.98%   +22.94%     
- Complexity             0      747      +747     
==================================================
  Files                 75       77        +2     
  Lines               8075     2619     -5456     
  Branches               0      231      +231     
==================================================
- Hits                4687     2121     -2566     
+ Misses              3031      357     -2674     
+ Partials             357      141      -216     
```

| Flag | Coverage Δ | |
|---|---|---|
| java-unittests | `80.98% <ø> (?)` | |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | Coverage Δ | |
|---|---|---|
| [control-plane/pkg/contract/extensions.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvZXh0ZW5zaW9ucy5nbw==) | | |
| [...ne/pkg/reconciler/trigger/namespaced\_controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL25hbWVzcGFjZWRfY29udHJvbGxlci5nbw==) | | |
| [control-plane/pkg/prober/async\_prober.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcHJvYmVyL2FzeW5jX3Byb2Jlci5nbw==) | | |
| [...ntrol-plane/pkg/reconciler/channel/v2/channelv2.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9jaGFubmVsL3YyL2NoYW5uZWx2Mi5nbw==) | | |
| [control-plane/pkg/security/secret.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvc2VjdXJpdHkvc2VjcmV0Lmdv) | | |
| [control-plane/pkg/contract/log\_marshaller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvY29udHJhY3QvbG9nX21hcnNoYWxsZXIuZ28=) | | |
| [control-plane/pkg/reconciler/trigger/controller.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci90cmlnZ2VyL2NvbnRyb2xsZXIuZ28=) | | |
| [...nternals/kafka/eventing/v1alpha1/consumer\_types.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9pbnRlcm5hbHMva2Fma2EvZXZlbnRpbmcvdjFhbHBoYTEvY29uc3VtZXJfdHlwZXMuZ28=) | | |
| [control-plane/pkg/reconciler/broker/broker.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvcmVjb25jaWxlci9icm9rZXIvYnJva2VyLmdv) | | |
| [...pkg/apis/eventing/v1alpha1/kafka\_sink\_lifecycle.go](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox#diff-Y29udHJvbC1wbGFuZS9wa2cvYXBpcy9ldmVudGluZy92MWFscGhhMS9rYWZrYV9zaW5rX2xpZmVjeWNsZS5nbw==) | | |
| ... and [142 more](https://codecov.io/gh/knative-sandbox/eventing-kafka-broker/pull/2826/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=knative-sandbox)

knative-prow[bot]: @knative-prow-robot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8 | b99ad684ebe06ae3044385127cba7a6e8cdfc0b1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2826/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8/1599788571122733056) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: /lgtm
/approve
/retest-required
knative-prow[bot]: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2826#issuecomment-1338808727" title="Approved">aliok</a>*, *<a href="https://github.com/knative-sandbox/eventing-kafka-broker/pull/2826#" title="Author self-approved">knative-prow-robot</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=knative-sandbox%2Feventing-kafka-broker).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/knative-sandbox/eventing-kafka-broker/blob/release-1.8/OWNERS)~~ [aliok]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
knative-prow[bot]: @knative-prow-robot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8 | b99ad684ebe06ae3044385127cba7a6e8cdfc0b1 | [link](https://prow.knative.dev/view/gs/knative-prow/pr-logs/pull/knative-sandbox_eventing-kafka-broker/2826/channel-integration-tests-sasl-plain_eventing-kafka-broker_release-1.8/1600007499149217792) | true | `/test channel-integration-tests-sasl-plain`

[Your PR dashboard](https://prow.knative.dev/pr).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
aliok: /test channel-integration-tests-sasl-plain

Patch:
diff --git a/vendor/github.com/manifestival/manifestival/fake/client.go b/vendor/github.com/manifestival/manifestival/fake/client.go
deleted file mode 100644
index 60e9c31be..000000000
--- a/vendor/github.com/manifestival/manifestival/fake/client.go
+++ /dev/null
@@ -1,102 +0,0 @@
-package fake
-
-import (
-	"fmt"
-
-	mf "github.com/manifestival/manifestival"
-	"k8s.io/apimachinery/pkg/api/errors"
-	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
-	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/runtime/schema"
-	"k8s.io/client-go/kubernetes/scheme"
-)
-
-var _ mf.Client = &Client{}
-
-// A convenient way to stub out a Client for test fixtures. Default
-// behavior does nothing and returns a nil error.
-type Client struct {
-	Stubs
-}
-
-// Override any of the Client functions
-type Stubs struct {
-	Create mutator
-	Update mutator
-	Delete mutator
-	Get    accessor
-}
-
-type mutator func(obj *unstructured.Unstructured) error
-type accessor func(obj *unstructured.Unstructured) (*unstructured.Unstructured, error)
-
-// New returns a fully-functioning Client, "persisting" resources in a
-// map, optionally initialized with some API objects
-func New(objs ...runtime.Object) Client {
-	store := map[string]*unstructured.Unstructured{}
-	key := func(u *unstructured.Unstructured) string {
-		return fmt.Sprintf("%s, %s/%s", u.GroupVersionKind(), u.GetNamespace(), u.GetName())
-	}
-	for _, obj := range objs {
-		u := &unstructured.Unstructured{}
-		if err := scheme.Scheme.Convert(obj, u, nil); err != nil {
-			panic(err)
-		}
-		store[key(u)] = u
-	}
-	apply := func(u *unstructured.Unstructured) error {
-		store[key(u)] = u
-		return nil
-	}
-	return Client{
-		Stubs{
-			Create: apply,
-			Update: apply,
-			Delete: func(u *unstructured.Unstructured) error {
-				delete(store, key(u))
-				return nil
-			},
-			Get: func(u *unstructured.Unstructured) (*unstructured.Unstructured, error) {
-				v, found := store[key(u)]
-				if !found {
-					gvk := u.GroupVersionKind()
-					gr := schema.GroupResource{Group: gvk.Group, Resource: gvk.Kind}
-					return nil, errors.NewNotFound(gr, u.GetName())
-				}
-				return v, nil
-			},
-		},
-	}
-}
-
-// Manifestival.Client.Create
-func (c Client) Create(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
-	if c.Stubs.Create != nil {
-		return c.Stubs.Create(obj)
-	}
-	return nil
-}
-
-// Manifestival.Client.Update
-func (c Client) Update(obj *unstructured.Unstructured, options ...mf.ApplyOption) error {
-	if c.Stubs.Update != nil {
-		return c.Stubs.Update(obj)
-	}
-	return nil
-}
-
-// Manifestival.Client.Delete
-func (c Client) Delete(obj *unstructured.Unstructured, options ...mf.DeleteOption) error {
-	if c.Stubs.Delete != nil {
-		return c.Stubs.Delete(obj)
-	}
-	return nil
-}
-
-// Manifestival.Client.Get
-func (c Client) Get(obj *unstructured.Unstructured) (*unstructured.Unstructured, error) {
-	if c.Stubs.Get != nil {
-		return c.Stubs.Get(obj)
-	}
-	return nil, nil
-}
diff --git a/vendor/modules.txt b/vendor/modules.txt
index 97de57bb8..d7b1dcf30 100644
--- a/vendor/modules.txt
+++ b/vendor/modules.txt
@@ -329,7 +329,6 @@ github.com/manifestival/client-go-client/pkg/dynamic
 # github.com/manifestival/manifestival v0.7.1
 ## explicit; go 1.15
 github.com/manifestival/manifestival
-github.com/manifestival/manifestival/fake
 github.com/manifestival/manifestival/internal/overlay
 github.com/manifestival/manifestival/internal/patch
 github.com/manifestival/manifestival/internal/sources


Test patch:
diff --git a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
index 7d8695076..1c344963d 100644
--- a/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
+++ b/control-plane/pkg/reconciler/broker/namespaced_broker_test.go
@@ -21,7 +21,9 @@ import (
 	"fmt"
 	"testing"
 
+	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/utils/pointer"
 	sources "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
 
@@ -32,15 +34,15 @@ import (
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober"
 	"knative.dev/eventing-kafka-broker/control-plane/pkg/prober/probertesting"
 
-	mffake "github.com/manifestival/manifestival/fake"
-
 	"github.com/Shopify/sarama"
+	"github.com/manifestival/client-go-client"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/types"
 	clientgotesting "k8s.io/client-go/testing"
 	eventing "knative.dev/eventing/pkg/apis/eventing/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	"knative.dev/pkg/controller"
+	dynamicclientfake "knative.dev/pkg/injection/clients/dynamicclient/fake"
 	"knative.dev/pkg/logging"
 	. "knative.dev/pkg/reconciler/testing"
 	"knative.dev/pkg/resolver"
@@ -101,7 +103,8 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 				NewServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
 				reconcilertesting.NewService("kafka-broker-ingress", SystemNamespace),
 				NewClusterRoleBinding("knative-kafka-broker-data-plane",
-					WithSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane", "knative-kafka-broker-data-plane"),
+					WithClusterRoleBindingSubjectServiceAccount(SystemNamespace, "knative-kafka-broker-data-plane"),
+					WithClusterRoleBindingRoleRef("knative-kafka-broker-data-plane"),
 				),
 			},
 			Key: testKey,
@@ -109,6 +112,56 @@ func namespacedBrokerReconciliation(t *testing.T, format string, env config.Env)
 				finalizerUpdatedEvent,
 			},
 			WantCreates: []runtime.Object{
+				ToManifestivalResource(t,
+					DataPlaneConfigMap(BrokerNamespace, env.DataPlaneConfigConfigMapName, ConsumerConfigKey,
+						DataPlaneConfigInitialOffset(ConsumerConfigKey, sources.OffsetLatest),
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewConfigMap(
+						"config-tracing",
+						BrokerNamespace,
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewConfigMap(
+						"kafka-config-logging",
+						BrokerNamespace,
+					),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewDeployment("kafka-broker-receiver", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewDeployment("kafka-broker-dispatcher", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					NewServiceAccount(BrokerNamespace, "knative-kafka-broker-data-plane"),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t,
+					reconcilertesting.NewService("kafka-broker-ingress", BrokerNamespace),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
+				ToManifestivalResource(t, NewRoleBinding(BrokerNamespace, "knative-kafka-broker-data-plane",
+					WithRoleBindingSubjectServiceAccount(BrokerNamespace, "knative-kafka-broker-data-plane"),
+					WithRoleBindingClusterRoleRef("knative-kafka-broker-data-plane"),
+				),
+					WithNamespacedBrokerOwnerRef,
+					WithNamespacedLabel,
+				),
 				NewConfigMapWithBinaryData(BrokerNamespace, env.ContractConfigMapName, nil),
 			},
 			WantUpdates: []clientgotesting.UpdateActionImpl{
@@ -264,7 +317,7 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 			proberMock = p.(prober.Prober)
 		}
 
-		mfcMockClient := mffake.New()
+		mfcMockClient, _ := client.NewUnsafeDynamicClient(dynamicclientfake.Get(ctx))
 
 		reconciler := &NamespacedReconciler{
 			Reconciler: &base.Reconciler{
@@ -318,3 +371,55 @@ func useTableNamespaced(t *testing.T, table TableTest, env *config.Env) {
 		return r
 	}))
 }
+
+func WithNamespacedBrokerOwnerRef(u *unstructured.Unstructured) {
+	refs := u.GetOwnerReferences()
+	if refs == nil {
+		refs = []metav1.OwnerReference{}
+	}
+	refs = append(refs, metav1.OwnerReference{
+		APIVersion:         eventing.SchemeGroupVersion.String(),
+		Kind:               "Broker",
+		Name:               BrokerName,
+		UID:                BrokerUUID,
+		Controller:         pointer.Bool(false),
+		BlockOwnerDeletion: pointer.Bool(true),
+	})
+	u.SetOwnerReferences(refs)
+}
+
+func WithNamespacedLabel(u *unstructured.Unstructured) {
+	labels := u.GetLabels()
+	if labels == nil {
+		labels = make(map[string]string)
+	}
+	labels[kafka.NamespacedBrokerDataplaneLabelKey] = kafka.NamespacedBrokerDataplaneLabelValue
+	u.SetLabels(labels)
+}
+
+func ToManifestivalResource(t *testing.T, obj runtime.Object, mutators ...UnstructuredMutator) runtime.Object {
+	m := func(u *unstructured.Unstructured) {
+		annotations := u.GetAnnotations()
+		if annotations == nil {
+			annotations = make(map[string]string)
+		}
+		annotations["manifestival"] = "new"
+		u.SetAnnotations(annotations)
+
+		annotations[corev1.LastAppliedConfigAnnotation] = lastApplied(u)
+		u.SetAnnotations(annotations)
+	}
+	mutators = append(mutators, m)
+	return ToUnstructured(t, obj, mutators...)
+}
+
+// lastApplied returns a JSON string denoting the resource's state
+func lastApplied(obj *unstructured.Unstructured) string {
+	ann := obj.GetAnnotations()
+	if len(ann) > 0 {
+		delete(ann, corev1.LastAppliedConfigAnnotation)
+		obj.SetAnnotations(ann)
+	}
+	bytes, _ := obj.MarshalJSON()
+	return string(bytes)
+}
diff --git a/control-plane/pkg/reconciler/testing/factory.go b/control-plane/pkg/reconciler/testing/factory.go
index 4e7061638..02e52d45c 100644
--- a/control-plane/pkg/reconciler/testing/factory.go
+++ b/control-plane/pkg/reconciler/testing/factory.go
@@ -66,7 +66,7 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 
 		ctx, dynamicClient := fakedynamicclient.With(ctx,
 			newScheme(),
-			ToUnstructured(t, row.Objects)...,
+			ToUnstructuredList(t, row.Objects)...,
 		)
 
 		dynamicScheme := runtime.NewScheme()
@@ -118,36 +118,48 @@ func NewFactory(env *config.Env, ctor Ctor) Factory {
 	}
 }
 
-// ToUnstructured takes a list of k8s resources and converts them to
+// ToUnstructuredList takes a list of k8s resources and converts them to
 // Unstructured objects.
 // We must pass objects as Unstructured to the dynamic client fake, or it
 // won't handle them properly.
-func ToUnstructured(t *testing.T, objs []runtime.Object) (us []runtime.Object) {
-	sch := newScheme()
+func ToUnstructuredList(t *testing.T, objs []runtime.Object) (us []runtime.Object) {
 	for _, obj := range objs {
-		obj = obj.DeepCopyObject() // Don't mess with the primary copy
-		// Determine and set the TypeMeta for this object based on our test scheme.
-		gvks, _, err := sch.ObjectKinds(obj)
-		if err != nil {
-			t.Fatalf("Unable to determine kind for type: %v", err)
-		}
-		apiv, k := gvks[0].ToAPIVersionAndKind()
-		ta, err := meta.TypeAccessor(obj)
-		if err != nil {
-			t.Fatalf("Unable to create type accessor: %v", err)
-		}
-		ta.SetAPIVersion(apiv)
-		ta.SetKind(k)
-
-		b, err := json.Marshal(obj)
-		if err != nil {
-			t.Fatalf("Unable to marshal: %v", err)
-		}
-		u := &unstructured.Unstructured{}
-		if err := json.Unmarshal(b, u); err != nil {
-			t.Fatalf("Unable to unmarshal: %v", err)
-		}
+		u := ToUnstructured(t, obj)
 		us = append(us, u)
 	}
 	return
 }
+
+type UnstructuredMutator func(u *unstructured.Unstructured)
+
+// ToUnstructured takes a single k8s resource and converts it to
+// Unstructured object.
+func ToUnstructured(t *testing.T, obj runtime.Object, mutators ...UnstructuredMutator) runtime.Object {
+	sch := newScheme()
+	obj = obj.DeepCopyObject() // Don't mess with the primary copy
+	// Determine and set the TypeMeta for this object based on our test scheme.
+	gvks, _, err := sch.ObjectKinds(obj)
+	if err != nil {
+		t.Fatalf("Unable to determine kind for type: %v", err)
+	}
+	apiv, k := gvks[0].ToAPIVersionAndKind()
+	ta, err := meta.TypeAccessor(obj)
+	if err != nil {
+		t.Fatalf("Unable to create type accessor: %v", err)
+	}
+	ta.SetAPIVersion(apiv)
+	ta.SetKind(k)
+
+	b, err := json.Marshal(obj)
+	if err != nil {
+		t.Fatalf("Unable to marshal: %v", err)
+	}
+	u := &unstructured.Unstructured{}
+	if err := json.Unmarshal(b, u); err != nil {
+		t.Fatalf("Unable to unmarshal: %v", err)
+	}
+	for _, m := range mutators {
+		m(u)
+	}
+	return u
+}
diff --git a/control-plane/pkg/reconciler/testing/objects_common.go b/control-plane/pkg/reconciler/testing/objects_common.go
index 5667432a1..29d87e3b3 100644
--- a/control-plane/pkg/reconciler/testing/objects_common.go
+++ b/control-plane/pkg/reconciler/testing/objects_common.go
@@ -158,7 +158,29 @@ func NewRoleBinding(namespace string, name string, mutations ...func(account *rb
 	return s
 }
 
-func WithSubjectServiceAccount(saNamespace, saName, roleName string) func(account *rbacv1.ClusterRoleBinding) {
+func WithRoleBindingSubjectServiceAccount(saNamespace, saName string) func(account *rbacv1.RoleBinding) {
+	return func(cb *rbacv1.RoleBinding) {
+		cb.Subjects = []rbacv1.Subject{
+			{
+				Kind:      "ServiceAccount",
+				Namespace: saNamespace,
+				Name:      saName,
+			},
+		}
+	}
+}
+
+func WithRoleBindingClusterRoleRef(roleName string) func(account *rbacv1.RoleBinding) {
+	return func(cb *rbacv1.RoleBinding) {
+		cb.RoleRef = rbacv1.RoleRef{
+			Kind:     "ClusterRole",
+			Name:     roleName,
+			APIGroup: "rbac.authorization.k8s.io",
+		}
+	}
+}
+
+func WithClusterRoleBindingSubjectServiceAccount(saNamespace, saName string) func(account *rbacv1.ClusterRoleBinding) {
 	return func(cb *rbacv1.ClusterRoleBinding) {
 		cb.Subjects = []rbacv1.Subject{
 			{
@@ -167,6 +189,11 @@ func WithSubjectServiceAccount(saNamespace, saName, roleName string) func(accoun
 				Name:      saName,
 			},
 		}
+	}
+}
+
+func WithClusterRoleBindingRoleRef(roleName string) func(account *rbacv1.ClusterRoleBinding) {
+	return func(cb *rbacv1.ClusterRoleBinding) {
 		cb.RoleRef = rbacv1.RoleRef{
 			Kind:     "ClusterRole",
 			Name:     roleName,
@@ -634,7 +661,7 @@ func PodAnnotations(annotations map[string]string) PodOption {
 	}
 }
 
-func DataPlaneConfigMap(namespace, name, key string, options ...base.ConfigMapOption) *corev1.ConfigMap {
+func DataPlaneConfigMap(namespace, name, key string, options ...reconcilertesting.ConfigMapOption) *corev1.ConfigMap {
 	cm := &corev1.ConfigMap{
 		ObjectMeta: metav1.ObjectMeta{
 			Namespace: namespace,
@@ -652,7 +679,7 @@ func DataPlaneConfigMap(namespace, name, key string, options ...base.ConfigMapOp
 	return cm
 }
 
-func DataPlaneConfigInitialOffset(key string, offset sources.Offset) base.ConfigMapOption {
+func DataPlaneConfigInitialOffset(key string, offset sources.Offset) reconcilertesting.ConfigMapOption {
 	return func(cm *corev1.ConfigMap) {
 		props := properties.MustLoadString(cm.Data[key])
 		_, _, _ = props.Set("auto.offset.reset", string(offset))