added unit test for kerberos config

Author: adwk67

rust/crd/src/listener.rs

@@ -267,8 +267,9 @@ mod tests {
         builder::meta::ObjectMetaBuilder,
         commons::{
             authentication::{
-                tls::AuthenticationProvider, AuthenticationClass, AuthenticationClassProvider,
-                AuthenticationClassSpec,
+                kerberos,
+                tls::{self},
+                AuthenticationClass, AuthenticationClassProvider, AuthenticationClassSpec,
             },
             networking::DomainName,
         },
@@ -307,7 +308,7 @@ mod tests {
             ResolvedAuthenticationClasses::new(vec![AuthenticationClass {
                 metadata: ObjectMetaBuilder::new().name("auth-class").build(),
                 spec: AuthenticationClassSpec {
-                    provider: AuthenticationClassProvider::Tls(AuthenticationProvider {
+                    provider: AuthenticationClassProvider::Tls(tls::AuthenticationProvider {
                         client_cert_secret_class: Some("client-auth-secret-class".to_string()),
                     }),
                 },
@@ -456,4 +457,97 @@ mod tests {
             )
         );
     }
+
+    #[test]
+    fn test_get_kafka_kerberos_listeners_config() {
+        let object_name = "simple-kafka-broker-default";
+        let cluster_info = default_cluster_info();
+
+        let kafka_cluster = r#"
+        apiVersion: kafka.stackable.tech/v1alpha1
+        kind: KafkaCluster
+        metadata:
+          name: simple-kafka
+          namespace: default
+        spec:
+          image:
+            productVersion: 3.7.1
+          clusterConfig:
+            authentication:
+              - authenticationClass: kafka-kerberos
+            tls:
+              serverSecretClass: tls
+            zookeeperConfigMapName: xyz
+        "#;
+        let kafka: KafkaCluster = serde_yaml::from_str(kafka_cluster).expect("illegal test input");
+        let kafka_security = KafkaTlsSecurity::new(
+            ResolvedAuthenticationClasses::new(vec![AuthenticationClass {
+                metadata: ObjectMetaBuilder::new().name("auth-class").build(),
+                spec: AuthenticationClassSpec {
+                    provider: AuthenticationClassProvider::Kerberos(
+                        kerberos::AuthenticationProvider {
+                            kerberos_secret_class: "kerberos-secret-class".to_string(),
+                        },
+                    ),
+                },
+            }]),
+            "tls".to_string(),
+            Some("tls".to_string()),
+        );
+
+        let config =
+            get_kafka_listener_config(&kafka, &kafka_security, object_name, &cluster_info).unwrap();
+
+        assert_eq!(
+            config.listeners(),
+            format!(
+                "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port},{bootstrap_name}://{bootstrap_host}:{bootstrap_port}",
+                name = KafkaListenerName::Client,
+                host = LISTENER_LOCAL_ADDRESS,
+                port = kafka_security.client_port(),
+                internal_name = KafkaListenerName::Internal,
+                internal_host = LISTENER_LOCAL_ADDRESS,
+                internal_port = kafka_security.internal_port(),
+                bootstrap_name = KafkaListenerName::Bootstrap,
+                bootstrap_host = LISTENER_LOCAL_ADDRESS,
+             bootstrap_port = kafka_security.bootstrap_port(),
+            )
+        );
+
+        assert_eq!(
+            config.advertised_listeners(),
+            format!(
+                "{name}://{host}:{port},{internal_name}://{internal_host}:{internal_port},{bootstrap_name}://{bootstrap_host}:{bootstrap_port}",
+                name = KafkaListenerName::Client,
+                host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+                port = node_port_cmd(
+                    STACKABLE_LISTENER_BROKER_DIR,
+                    kafka_security.client_port_name()
+                ),
+                internal_name = KafkaListenerName::Internal,
+                internal_host = pod_fqdn(&kafka, object_name, &cluster_info).unwrap(),
+                internal_port = kafka_security.internal_port(),
+                bootstrap_name = KafkaListenerName::Bootstrap,
+                bootstrap_host = node_address_cmd(STACKABLE_LISTENER_BROKER_DIR),
+                bootstrap_port = node_port_cmd(
+                    STACKABLE_LISTENER_BROKER_DIR,
+                    kafka_security.client_port_name()
+                ),
+            )
+        );
+
+        assert_eq!(
+            config.listener_security_protocol_map(),
+            format!(
+                "{name}:{protocol},{internal_name}:{internal_protocol},{bootstrap_name}:{bootstrap_protocol}",
+                name = KafkaListenerName::Client,
+                protocol = KafkaListenerProtocol::SaslSsl,
+                internal_name = KafkaListenerName::Internal,
+                internal_protocol = KafkaListenerProtocol::Ssl,
+                bootstrap_name = KafkaListenerName::Bootstrap,
+                bootstrap_protocol = KafkaListenerProtocol::SaslSsl,
+
+            )
+        );
+    }
 }