test: cover OAUTHBEARER passthrough getters and Go token provider helpers

Adds unit tests for the previously uncovered patch lines flagged by Codecov:
- KubernetesCredentials SASLJaasConfig/SASLLoginCallbackHandlerClass (decode,
  absent, null secret, null data)
- Credentials default methods returning null via a minimal implementation
- Go HasTokenProvider (common.go) and UnsupportedTokenProvider/errorTokenProvider
  (token_provider.go)

Author: Damien Metzler

control-plane/pkg/security/oauth/common_test.go

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2025 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 oauth
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestHasTokenProvider(t *testing.T) {
+	tests := []struct {
+		name string
+		data map[string][]byte
+		want bool
+	}{
+		{
+			name: "tokenProvider present",
+			data: map[string][]byte{saslTokenProviderKey: []byte(mskAccessTokenProvider)},
+			want: true,
+		},
+		{
+			name: "tokenProvider present but empty still counts as set",
+			data: map[string][]byte{saslTokenProviderKey: []byte("")},
+			want: true,
+		},
+		{
+			name: "tokenProvider absent",
+			data: map[string][]byte{saslAWSRegion: []byte(defaultAWSRegion)},
+			want: false,
+		},
+		{
+			name: "empty data",
+			data: map[string][]byte{},
+			want: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			assert.Equal(t, tt.want, HasTokenProvider(tt.data))
+		})
+	}
+}

control-plane/pkg/security/oauth/token_provider_test.go

@@ -139,3 +139,15 @@ func TestNewTokenProvider(t *testing.T) {
 		})
 	}
 }
+
+func TestUnsupportedTokenProvider(t *testing.T) {
+	const msg = "OAUTHBEARER without a Go token provider is not supported by the control plane"
+
+	provider := UnsupportedTokenProvider(msg)
+	assert.NotNil(t, provider)
+
+	token, err := provider.Token()
+	assert.Nil(t, token)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), msg)
+}

data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/CredentialsTest.java

@@ -0,0 +1,80 @@
+/*
+ * Copyright © 2018 Knative Authors (knative-dev@googlegroups.com)
+ *
+ * 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 dev.knative.eventing.kafka.broker.core.security;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.apache.kafka.common.security.auth.SecurityProtocol;
+import org.junit.jupiter.api.Test;
+
+public class CredentialsTest {
+
+    /**
+     * A minimal {@link Credentials} implementation that does not override the
+     * OAUTHBEARER passthrough default methods, exercising their default (null) behaviour.
+     */
+    private static final class MinimalCredentials implements Credentials {
+        @Override
+        public String caCertificates() {
+            return null;
+        }
+
+        @Override
+        public String userCertificate() {
+            return null;
+        }
+
+        @Override
+        public String userKey() {
+            return null;
+        }
+
+        @Override
+        public String SASLMechanism() {
+            return null;
+        }
+
+        @Override
+        public String SASLUsername() {
+            return null;
+        }
+
+        @Override
+        public String SASLPassword() {
+            return null;
+        }
+
+        @Override
+        public SecurityProtocol securityProtocol() {
+            return null;
+        }
+
+        @Override
+        public boolean skipClientAuth() {
+            return false;
+        }
+    }
+
+    @Test
+    public void defaultSASLJaasConfigReturnsNull() {
+        assertThat(new MinimalCredentials().SASLJaasConfig()).isNull();
+    }
+
+    @Test
+    public void defaultSASLLoginCallbackHandlerClassReturnsNull() {
+        assertThat(new MinimalCredentials().SASLLoginCallbackHandlerClass()).isNull();
+    }
+}

data-plane/core/src/test/java/dev/knative/eventing/kafka/broker/core/security/KubernetesCredentialsTest.java

@@ -96,6 +96,53 @@ public void getKubernetesCredentialsFromNullSecretData() {
         assertAll(new HashMap<>(), credentials);
     }
 
+    @Test
+    public void saslJaasConfigAndCallbackHandlerAreDecoded() {
+        final var data = Map.of(
+                KubernetesCredentials.SASL_MECHANISM, "OAUTHBEARER",
+                KubernetesCredentials.SASL_JAAS_CONFIG_KEY,
+                        "org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;",
+                KubernetesCredentials.SASL_LOGIN_CALLBACK_HANDLER_CLASS_KEY, "com.example.MyCallbackHandler");
+
+        final var credentials = getKubernetesCredentialsFromSecretData(data);
+
+        assertThat(credentials.SASLJaasConfig())
+                .isEqualTo("org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required;");
+        assertThat(credentials.SASLLoginCallbackHandlerClass()).isEqualTo("com.example.MyCallbackHandler");
+    }
+
+    @Test
+    public void saslJaasConfigAndCallbackHandlerAbsentReturnNull() {
+        final var data = Map.of(KubernetesCredentials.SASL_MECHANISM, "OAUTHBEARER");
+
+        final var credentials = getKubernetesCredentialsFromSecretData(data);
+
+        assertThat(credentials.SASLJaasConfig()).isNull();
+        assertThat(credentials.SASLLoginCallbackHandlerClass()).isNull();
+    }
+
+    @Test
+    public void saslJaasConfigAndCallbackHandlerNullOnNullSecret() {
+        final Secret secret = null;
+        final var credentials = new KubernetesCredentials(secret);
+
+        assertThat(credentials.SASLJaasConfig()).isNull();
+        assertThat(credentials.SASLLoginCallbackHandlerClass()).isNull();
+    }
+
+    @Test
+    public void saslJaasConfigAndCallbackHandlerNullOnNullSecretData() {
+        final var credentials = new KubernetesCredentials(new SecretBuilder()
+                .withNewMetadata()
+                .withNamespace("ns1")
+                .withName("name1")
+                .endMetadata()
+                .build());
+
+        assertThat(credentials.SASLJaasConfig()).isNull();
+        assertThat(credentials.SASLLoginCallbackHandlerClass()).isNull();
+    }
+
     @Test
     public void unknownSecurityProtocolReturnsNull() {
         final var data = Map.of(KubernetesCredentials.SECURITY_PROTOCOL, "SASSO_PLAINTEXT");