Fixed unit tests and build

Signed-off-by: Calum Murray <[email protected]>
knative-extensions · Oct 11, 2023 · c8f2136 · c8f2136
1 parent 216ca26
commit c8f2136
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 2 deletions.
diff --git a/pkg/source/adapter/adapter.go b/pkg/source/adapter/adapter.go
@@ -38,6 +38,7 @@ import (
 	"knative.dev/pkg/logging"
 
 	"knative.dev/eventing/pkg/adapter/v2"
+	"knative.dev/eventing/pkg/auth"
 	"knative.dev/eventing/pkg/kncloudevents"
 
 	"knative.dev/eventing-kafka/pkg/common/consumer"
@@ -75,6 +76,7 @@ type Adapter struct {
 	keyTypeMapper func([]byte) interface{}
 	rateLimiter   *rate.Limiter
 	extensions    map[string]string
+	dispatcher    *kncloudevents.Dispatcher
 }
 
 var (
@@ -89,12 +91,15 @@ func NewAdapter(ctx context.Context, processed adapter.EnvConfigAccessor, sink d
 	logger := logging.FromContext(ctx)
 	config := processed.(*AdapterConfig)
 
+	oidcTokenProvider := auth.NewOIDCTokenProvider(ctx)
+
 	return &Adapter{
 		config:        config,
 		sink:          sink,
 		reporter:      reporter,
 		logger:        logger,
 		keyTypeMapper: getKeyTypeMapper(config.KeyType),
+		dispatcher:    kncloudevents.NewDispatcher(oidcTokenProvider),
 	}
 }
 func (a *Adapter) GetConsumerGroup() string {
@@ -176,7 +181,7 @@ func (a *Adapter) Handle(ctx context.Context, msg *sarama.ConsumerMessage) (bool
 		return false, fmt.Errorf("failed to get cloud event from consumer message: %w", err)
 	}
 
-	dispatchInfo, err := kncloudevents.SendEvent(ctx, *event, a.sink,
+	dispatchInfo, err := a.dispatcher.SendEvent(ctx, *event, a.sink,
 		kncloudevents.WithRetryConfig(retryConfig),
 		kncloudevents.WithTransformers(extensionAsTransformer(a.extensions)))
 	if err != nil {

diff --git a/pkg/source/adapter/adapter_test.go b/pkg/source/adapter/adapter_test.go
@@ -35,6 +35,8 @@ import (
 	duckv1 "knative.dev/pkg/apis/duck/v1"
 
 	sourcesv1beta1 "knative.dev/eventing-kafka/pkg/apis/sources/v1beta1"
+
+	fakekubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 )
 
 func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
@@ -342,6 +344,8 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 				URL: sinkURI,
 			}
 
+			ctx, _ := fakekubeclient.With(context.Background())
+
 			a := &Adapter{
 				config: &AdapterConfig{
 					EnvConfig: adapter.EnvConfig{
@@ -358,7 +362,7 @@ func TestPostMessage_ServeHTTP_binary_mode(t *testing.T) {
 				keyTypeMapper: getKeyTypeMapper(tc.keyTypeMapper),
 			}
 
-			_, err = a.Handle(context.TODO(), tc.message)
+			_, err = a.Handle(ctx, tc.message)
 			if tc.error && err == nil {
 				t.Errorf("expected error, but got %v", err)
 			}
@@ -434,6 +438,7 @@ func sinkRejected(writer http.ResponseWriter, _ *http.Request) {
 
 func TestAdapter_Start(t *testing.T) { // just increase code coverage
 	ctx, cancel := context.WithCancel(context.Background())
+	ctx, _ = fakekubeclient.With(ctx)
 
 	// Increasing coverage
 	_ = os.Setenv("KAFKA_BOOTSTRAP_SERVERS", "my-cluster-kafka-bootstrap.my-kafka-namespace:9092")