/*
 * Copyright Kroxylicious Authors.
 *
 * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0
 */

package io.kroxylicious.kubernetes.operator;

import java.time.Clock;
import java.time.Instant;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.function.Consumer;
import java.util.stream.Stream;

import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
import org.junit.jupiter.params.provider.MethodSource;
import org.mockito.junit.jupiter.MockitoExtension;

import io.fabric8.kubernetes.api.model.APIGroup;
import io.fabric8.kubernetes.api.model.ConfigMap;
import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
import io.fabric8.kubernetes.api.model.KubernetesResourceList;
import io.fabric8.kubernetes.api.model.Secret;
import io.fabric8.kubernetes.api.model.SecretBuilder;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.MixedOperation;
import io.fabric8.kubernetes.client.dsl.Resource;
import io.fabric8.kubernetes.client.server.mock.EnableKubernetesMockClient;
import io.javaoperatorsdk.operator.api.reconciler.Context;
import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext;
import io.javaoperatorsdk.operator.api.reconciler.UpdateControl;
import io.javaoperatorsdk.operator.processing.event.ResourceID;
import io.strimzi.api.kafka.model.kafka.Kafka;
import io.strimzi.api.kafka.model.kafka.KafkaBuilder;
import io.strimzi.api.kafka.model.kafka.listener.GenericKafkaListenerBuilder;
import io.strimzi.api.kafka.model.kafka.listener.ListenerAddressBuilder;
import io.strimzi.api.kafka.model.kafka.listener.ListenerStatusBuilder;

import io.kroxylicious.kubernetes.api.common.Condition;
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaService;
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceBuilder;
import io.kroxylicious.kubernetes.operator.assertj.ConditionListAssert;
import io.kroxylicious.kubernetes.operator.assertj.OperatorAssertions;

import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;

@EnableKubernetesMockClient(crud = true)
@ExtendWith(MockitoExtension.class)
class KafkaServiceReconcilerTest {

    public static final Clock TEST_CLOCK = Clock.fixed(Instant.EPOCH, ZoneId.of("Z"));

    public static final long OBSERVED_GENERATION = 1345L;
    public static final String KAFKA_GROUP_NAME = "kafka.strimzi.io";

    // @formatter:off
    public static final KafkaService SERVICE = new KafkaServiceBuilder()
            .withNewMetadata()
                .withName("test-service")
                .withGeneration(OBSERVED_GENERATION)
            .endMetadata()
            .withNewSpec()
                .withNewStrimziKafkaRef()
                    .withNewRef()
                        .withName("my-cluster")
                        .withKind("Kafka")
                        .withGroup("kafka.strimzi.io")
                    .endRef()
                .endStrimziKafkaRef()
                .withNewTls()
                    .withNewCertificateRef()
                        .withName("my-secret")
                    .endCertificateRef()
                    .withNewTrustAnchorRef()
                        .withNewRef()
                            .withName("my-configmap")
                        .endRef()
                    .endTrustAnchorRef()
                .endTls()
            .endSpec()
            .build();

    public static final Secret TLS_SECRET = new SecretBuilder()
            .withNewMetadata()
                .withName("my-secret")
                .withUid("uid")
                .withResourceVersion("6744")
            .endMetadata()
            .withType("kubernetes.io/tls")
            .addToData("tls.crt", "value")
            .addToData("tls.key", "value")
            .build();

    public static final Secret UNSUPPORTED_SECRET = new SecretBuilder()
            .withNewMetadata()
                .withName("my-secret")
                .withUid("uid")
                .withResourceVersion("7742")
            .endMetadata()
            .addToData("key", "value")
            .build();

    public static final ConfigMap PEM_CONFIG_MAP = new ConfigMapBuilder()
            .withNewMetadata()
                .withName("my-configmap")
                .withUid("uid")
                .withResourceVersion("7782")
            .endMetadata()
            .addToData("ca-bundle.pem", "value")
            .build();

    public static final Kafka KAFKA = new KafkaBuilder()
            .withNewMetadata()
            .withName("my-cluster")
            .withUid("uid")
            .withResourceVersion("7782")
            .endMetadata()
            .withNewSpec()
                .withNewKafka()
                    .withListeners(new GenericKafkaListenerBuilder()
                            .withName("plain")
                            .withTls(false)
                            .build())
                .endKafka()
            .endSpec()
            .withNewStatus()
            .withListeners(List.of(new ListenerStatusBuilder()
                            .withName("plain")
                            .withAddresses(new ListenerAddressBuilder()
                                    .withPort(8080)
                                    .withHost("localhost")
                                    .build())
                            .build()))
            .endStatus()
            .build();

    public static final Kafka UNSUPPORTED_KAFKA = new KafkaBuilder()
            .withNewMetadata()
            .withName("my-cluster")
            .withUid("uid")
            .withResourceVersion("7782")
            .endMetadata()
            .withNewSpec()
                .withNewKafka()
                    .withListeners(new GenericKafkaListenerBuilder()
                            .withName("tls")
                            .withTls(false)
                        .build())
                .endKafka()
            .endSpec()
            .withNewStatus()
            .withListeners(List.of(new ListenerStatusBuilder()
                    .withName("tls")
                    .withAddresses(new ListenerAddressBuilder()
                            .withPort(8080)
                            .withHost("localhost")
                            .build())
                    .build()))
            .endStatus()
            .build();

    public static final ConfigMap P12_CONFIG_MAP = new ConfigMapBuilder()
            .withNewMetadata()
                .withName("my-configmap")
                .withUid("uid")
                .withResourceVersion("3342")
            .endMetadata()
            .addToData("ca-bundle.p12", "value")
            .build();

    public static final ConfigMap JKS_CONFIG_MAP = new ConfigMapBuilder()
            .withNewMetadata()
                .withName("my-configmap")
                .withUid("uid")
                .withResourceVersion("1266")
            .endMetadata()
            .addToData("ca-bundle.jks", "value")
            .build();

    public static final ConfigMap UNSUPPORTED_CONFIG_MAP = new ConfigMapBuilder()
            .withNewMetadata()
                .withName("my-configmap")
                .withUid("uid")
                .withResourceVersion("8982")
            .endMetadata()
            .addToData("unsuppor.ted", "value")
            .build();

    // @formatter:on

    private KafkaServiceReconciler kafkaServiceReconciler;

    @BeforeEach
    void setUp() {
        kafkaServiceReconciler = new KafkaServiceReconciler(Clock.systemUTC());
    }

    @Test
    void shouldSetResolvedRefsToUnknown() {
        // given
        final KafkaService kafkaService = new KafkaServiceBuilder().withNewMetadata().withGeneration(OBSERVED_GENERATION).endMetadata().build();

        var reconciler = new KafkaServiceReconciler(TEST_CLOCK);

        Context<KafkaService> context = mock();

        // when
        var update = reconciler.updateErrorStatus(kafkaService, context, new RuntimeException("Boom!"));

        // then
        assertThat(update).isNotNull();
        assertThat(update.getResource()).isPresent();
        OperatorAssertions.assertThat(update.getResource().get().getStatus())
                .hasObservedGenerationInSyncWithMetadataOf(kafkaService)
                .singleCondition()
                .hasObservedGenerationInSyncWithMetadataOf(kafkaService)
                .isResolvedRefsUnknown("java.lang.RuntimeException", "Boom!")
                .hasLastTransitionTime(TEST_CLOCK.instant());

    }

    static List<Arguments> shouldSetResolvedRefs() {
        List<Arguments> result = new ArrayList<>();
        // no client cert, no trust
        {
            Context<KafkaService> context = mock();
            mockGetKafka(context, Optional.empty());
            mockGetSecret(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("no tls",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).withTls(null).endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsTrue()));
        }

        //
        // trust bundle cases....

        // no client cert, dangling trust bundle
        {
            Context<KafkaService> context = mock();
            mockGetKafka(context, Optional.empty());
            mockGetSecret(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("dangling trust bundle",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withCertificateRef(null).endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_REFS_NOT_FOUND,
                                    "spec.tls.trustAnchorRef: referenced configmap not found")));
        }

        // no client cert, unsupported trust bundle kind
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.of(UNSUPPORTED_CONFIG_MAP));
            result.add(Arguments.argumentSet("unsupported trustAnchorRef kind",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null)
                            .editTls()
                            .withCertificateRef(null)
                            .editTrustAnchorRef()
                            .editRef()
                            .withKind("Unsupported")
                            .endRef()
                            .endTrustAnchorRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_REF_GROUP_KIND_NOT_SUPPORTED,
                                    "spec.tls.trustAnchorRef supports referents: configmaps")));
        }

        // no client cert, trust bundle ref missing key
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.of(UNSUPPORTED_CONFIG_MAP));
            result.add(Arguments.argumentSet("trust bundle ref missing key",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withCertificateRef(null)
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_INVALID,
                                    "spec.tls.trustAnchorRef must specify 'key'")));
        }

        // missing listener name
        {
            Context<KafkaService> context = mock();
            KubernetesClient client = mock();
            when(context.getClient()).thenReturn(client);
            when(context.getClient().getApiGroup(KAFKA_GROUP_NAME)).thenReturn(new APIGroup());
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.of(KAFKA));
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("strimziKafkaRef missing listener name",
                    new KafkaServiceBuilder(SERVICE).editSpec()
                            .editStrimziKafkaRef()
                            .editRef()
                            .withGroup("kafka.strimzi.io")
                            .withKind("Kafka")
                            .withName("my-cluster")
                            .endRef()
                            .withListenerName("")
                            .endStrimziKafkaRef()
                            .endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_INVALID_REFERENCED_RESOURCE,
                                    "Referenced resource does not contain listener name: ")));
        }

        // unsupported kind
        {
            Context<KafkaService> context = mock();
            KubernetesClient client = mock();
            when(context.getClient()).thenReturn(client);
            when(context.getClient().getApiGroup(KAFKA_GROUP_NAME)).thenReturn(new APIGroup());
            mockGetSecret(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            mockGetKafka(context, Optional.of(KAFKA));
            result.add(Arguments.argumentSet("unsupported strimziKafkaRef kind",
                    new KafkaServiceBuilder(SERVICE).editSpec()
                            .editStrimziKafkaRef()
                            .editRef()
                            .withKind("Unsupported")
                            .endRef()
                            .withListenerName("plain")
                            .endStrimziKafkaRef()
                            .endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_REF_GROUP_KIND_NOT_SUPPORTED,
                                    "spec.strimziKafkaRef supports referents: kafka")));
        }

        // listener list is empty
        {
            Context<KafkaService> context = mock();
            KubernetesClient client = mock();
            when(context.getClient()).thenReturn(client);
            when(context.getClient().getApiGroup(KAFKA_GROUP_NAME)).thenReturn(new APIGroup());
            mockGetSecret(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            mockGetKafka(context, Optional.of(UNSUPPORTED_KAFKA));
            result.add(Arguments.argumentSet("listeners not present",
                    new KafkaServiceBuilder(SERVICE).editSpec()
                            .editStrimziKafkaRef()
                            .editRef()
                            .withGroup("kafka.strimzi.io")
                            .withKind("Kafka")
                            .withName("my-cluster")
                            .endRef()
                            .withListenerName("plain")
                            .endStrimziKafkaRef()
                            .endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_INVALID_REFERENCED_RESOURCE,
                                    "Referenced resource does not contain listener name: plain")));
        }

        // no client cert, unsupported trust bundle content
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.of(KAFKA));
            mockGetConfigMap(context, Optional.of(UNSUPPORTED_CONFIG_MAP));
            result.add(Arguments.argumentSet("unsupported trust bundle contents",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withCertificateRef(null)
                            .editTrustAnchorRef().withKey("ca-bundle.bob").endTrustAnchorRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_INVALID,
                                    "spec.tls.trustAnchorRef.key should end with .pem, .p12 or .jks")));
        }

        // no client cert, pem trust bundle
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.of(KAFKA));
            mockGetConfigMap(context, Optional.of(PEM_CONFIG_MAP));
            result.add(Arguments.argumentSet("pem trust bundle",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withCertificateRef(null)
                            .editTrustAnchorRef().withKey("ca-bundle.pem").endTrustAnchorRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsTrue()));
        }
        // no client cert, p12 trust bundle
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.of(P12_CONFIG_MAP));
            result.add(Arguments.argumentSet("p12 trust bundle",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withCertificateRef(null)
                            .editTrustAnchorRef().withKey("ca-bundle.p12").endTrustAnchorRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsTrue()));
        }
        // no client cert, jks trust bundle
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.of(JKS_CONFIG_MAP));
            result.add(Arguments.argumentSet("jks trust bundle",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withCertificateRef(null)
                            .editTrustAnchorRef().withKey("ca-bundle.jks").endTrustAnchorRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsTrue()));
        }

        //
        // client certificate cases....

        // dangling client cert, no trust
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("dangling client cert",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withTrustAnchorRef(null).endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_REFS_NOT_FOUND,
                                    "spec.tls.certificateRef: referenced secret not found")));
        }

        // unsupported client cert kind, no trust
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.empty());
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("unsupported client cert kind",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withTrustAnchorRef(null)
                            .editCertificateRef().withKind("Unsupported").endCertificateRef()
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_REF_GROUP_KIND_NOT_SUPPORTED,
                                    "spec.tls.certificateRef: supports referents: secrets")));
        }

        // unsupported client cert in Secret, no trust
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.of(UNSUPPORTED_SECRET));
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("unsupported client cert Secret content",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls()
                            .withTrustAnchorRef(null)
                            .endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsFalse(
                                    Condition.REASON_INVALID_REFERENCED_RESOURCE,
                                    "spec.tls.certificateRef: referenced secret should have 'type: kubernetes.io/tls'")));
        }

        // tls client cert, no trust
        {
            Context<KafkaService> context = mock();
            mockGetSecret(context, Optional.of(TLS_SECRET));
            mockGetKafka(context, Optional.empty());
            mockGetConfigMap(context, Optional.empty());
            result.add(Arguments.argumentSet("tls client cert",
                    new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withTrustAnchorRef(null).endTls().endSpec().build(),
                    context,
                    (Consumer<ConditionListAssert>) conditionList -> conditionList
                            .singleElement()
                            .isResolvedRefsTrue()));
        }

        return result;
    }

    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
    private static void mockGetConfigMap(
                                         Context<KafkaService> context,
                                         Optional<ConfigMap> empty) {
        when(context.getSecondaryResource(ConfigMap.class, KafkaServiceReconciler.CONFIG_MAPS_EVENT_SOURCE_NAME)).thenReturn(empty);
    }

    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
    private static void mockGetKafka(
                                     Context<KafkaService> context,
                                     Optional<Kafka> optional) {
        when(context.getSecondaryResource(Kafka.class, KafkaServiceReconciler.STRIMZI_KAFKA_EVENT_SOURCE_NAME)).thenReturn(optional);
    }

    @SuppressWarnings("OptionalUsedAsFieldOrParameterType")
    private static void mockGetSecret(
                                      Context<KafkaService> context,
                                      Optional<Secret> optional) {
        when(context.getSecondaryResource(Secret.class, KafkaServiceReconciler.SECRETS_EVENT_SOURCE_NAME)).thenReturn(optional);
    }

    @ParameterizedTest
    @MethodSource
    void shouldSetResolvedRefs(KafkaService kafkaService, Context<KafkaService> context, Consumer<ConditionListAssert> asserter) {

        // When
        final UpdateControl<KafkaService> updateControl = kafkaServiceReconciler.reconcile(kafkaService, context);

        // Then
        assertThat(updateControl).isNotNull();
        assertThat(updateControl.getResource()).isPresent();
        var c = OperatorAssertions.assertThat(updateControl.getResource().get().getStatus())
                .hasObservedGenerationInSyncWithMetadataOf(kafkaService)
                .conditionList();
        asserter.accept(c);
    }

    @Test
    void shouldSetReferentAnnotationWhenCertificateRefSecretPresent() {
        Context<KafkaService> context = mockContext();
        mockGetSecret(context, Optional.of(TLS_SECRET));
        KafkaService service = new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withTrustAnchorRef(null).endTls().endSpec().build();
        // When
        final UpdateControl<KafkaService> updateControl = kafkaServiceReconciler.reconcile(service, context);

        // Then
        assertThat(updateControl).isNotNull();
        assertThat(updateControl.isPatchResourceAndStatus()).isTrue();
        assertThat(updateControl.getResource()).isPresent().get().satisfies(kafkaService -> {
            assertThat(kafkaService.getMetadata().getAnnotations()).containsKey(Annotations.REFERENT_CHECKSUM_ANNOTATION_KEY);
        });
    }

    @Test
    void shouldSetReferentAnnotationWhenTrustAnchorRefConfigMapPresent() {
        Context<KafkaService> context = mockContext();
        mockGetConfigMap(context, Optional.of(PEM_CONFIG_MAP));
        KafkaService service = new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withCertificateRef(null).editTrustAnchorRef()
                .withKey("ca-bundle.pem")
                .endTrustAnchorRef().endTls().endSpec().build();
        // When
        final UpdateControl<KafkaService> updateControl = kafkaServiceReconciler.reconcile(service, context);

        // Then
        assertThat(updateControl).isNotNull();
        assertThat(updateControl.isPatchResourceAndStatus()).isTrue();
        assertThat(updateControl.getResource()).isPresent().get().satisfies(kafkaService -> {
            assertThat(kafkaService.getMetadata().getAnnotations()).containsKey(Annotations.REFERENT_CHECKSUM_ANNOTATION_KEY);
        });
    }

    @Test
    void shouldNotSetReferentAnnotationWhenServiceHasNoReferents() {
        Context<KafkaService> context = mockContext();
        KafkaService service = new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).editTls().withTrustAnchorRef(null).endTls().endSpec().build();
        // When
        final UpdateControl<KafkaService> updateControl = kafkaServiceReconciler.reconcile(service, context);

        // Then
        assertThat(updateControl).isNotNull();
        assertThat(updateControl.isPatchResourceAndStatus()).isTrue();
        assertThat(updateControl.getResource()).isPresent().get().satisfies(kafkaService -> {
            assertThat(kafkaService.getMetadata().getAnnotations()).doesNotContainKey(Annotations.REFERENT_CHECKSUM_ANNOTATION_KEY);
        });
    }

    @Test
    void canMapFromKafkaServiceWithTrustAnchorToConfigMap() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToConfigMap();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(SERVICE);

        // Then
        assertThat(secondaryResourceIDs).containsExactly(ResourceID.fromResource(PEM_CONFIG_MAP));
    }

    @Test
    void canMapFromKafkaServiceWithStrimziKafkaRefToStrimziKafka() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToStrimziKafka();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(SERVICE);

        // Then
        assertThat(secondaryResourceIDs).containsExactly(ResourceID.fromResource(KAFKA));
    }

    @Test
    void canMapFromKafkaServiceWithoutStrimziKafkaRefToStrimziKafka() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToStrimziKafka();
        var serviceNoStrimziKafkaRef = new KafkaServiceBuilder(SERVICE).editSpec().withStrimziKafkaRef(null).endSpec().build();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(serviceNoStrimziKafkaRef);

        // Then
        assertThat(secondaryResourceIDs).isEmpty();
    }

    @Test
    void canMapFromStrimziKafkaToKafkaService() {
        // Given
        EventSourceContext<KafkaService> eventSourceContext = mock();
        KubernetesClient client = mock();
        when(eventSourceContext.getClient()).thenReturn(client);

        KubernetesResourceList<KafkaService> mockList = mockKafkaServiceListOperation(client);
        when(mockList.getItems()).thenReturn(List.of(SERVICE));

        var mapper = KafkaServiceReconciler.strimziKafkaToKafkaService(eventSourceContext);

        // When
        var primaryResourceIDs = mapper.toPrimaryResourceIDs(KAFKA);

        // Then
        assertThat(primaryResourceIDs).containsExactly(ResourceID.fromResource(SERVICE));
    }

    @Test
    void canMapFromKafkaServiceWithoutTrustAnchorToConfigMap() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToConfigMap();
        var serviceNoTrustAnchor = new KafkaServiceBuilder(SERVICE).editSpec().editTls().withTrustAnchorRef(null).endTls().endSpec().build();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(serviceNoTrustAnchor);

        // Then
        assertThat(secondaryResourceIDs).isEmpty();
    }

    @Test
    void canMapFromConfigMapToKafkaService() {
        // Given
        EventSourceContext<KafkaService> eventSourceContext = mock();
        KubernetesClient client = mock();
        when(eventSourceContext.getClient()).thenReturn(client);

        KubernetesResourceList<KafkaService> mockList = mockKafkaServiceListOperation(client);
        when(mockList.getItems()).thenReturn(List.of(SERVICE));

        var mapper = KafkaServiceReconciler.configMapToKafkaService(eventSourceContext);

        // When
        var primaryResourceIDs = mapper.toPrimaryResourceIDs(PEM_CONFIG_MAP);

        // Then
        assertThat(primaryResourceIDs).containsExactly(ResourceID.fromResource(SERVICE));
    }

    static Stream<Arguments> mappingToConfigMapToleratesKafkaServicesWithoutTls() {
        return Stream.of(
                Arguments.argumentSet("without tls", new KafkaServiceBuilder(SERVICE).editSpec().withTls(null).endSpec().build()),
                Arguments.argumentSet("with tls but without trust anchor",
                        new KafkaServiceBuilder(SERVICE).editSpec().editTls().withTrustAnchorRef(null).endTls().endSpec().build()));
    }

    @ParameterizedTest
    @MethodSource
    void mappingToConfigMapToleratesKafkaServicesWithoutTls(KafkaService service) {
        // Given
        EventSourceContext<KafkaService> eventSourceContext = mock();
        KubernetesClient client = mock();
        when(eventSourceContext.getClient()).thenReturn(client);

        KubernetesResourceList<KafkaService> mockList = mockKafkaServiceListOperation(client);
        when(mockList.getItems()).thenReturn(List.of(service));

        // When
        var mapper = KafkaServiceReconciler.configMapToKafkaService(eventSourceContext);

        // Then
        var primaryResourceIDs = mapper.toPrimaryResourceIDs(new ConfigMapBuilder().withNewMetadata().withName("cm").endMetadata().build());
        assertThat(primaryResourceIDs).isEmpty();
    }

    @Test
    void canMapFromKafkaServiceWithClientCertToSecret() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToSecret();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(SERVICE);

        // Then
        assertThat(secondaryResourceIDs).containsExactly(ResourceID.fromResource(TLS_SECRET));
    }

    @Test
    void canMapFromKafkaServiceWithoutClientCertToSecret() {
        // Given
        var mapper = KafkaServiceReconciler.kafkaServiceToSecret();
        var serviceNoCert = new KafkaServiceBuilder(SERVICE).editSpec().editTls().withCertificateRef(null).endTls().endSpec().build();

        // When
        var secondaryResourceIDs = mapper.toSecondaryResourceIDs(serviceNoCert);

        // Then
        assertThat(secondaryResourceIDs).isEmpty();
    }

    static Stream<Arguments> mappingToSecretToleratesKafkaServicesWithoutTls() {
        return Stream.of(
                Arguments.argumentSet("without tls", new KafkaServiceBuilder(SERVICE).editSpec().withTls(null).endSpec().build()),
                Arguments.argumentSet("with tls but without client cert",
                        new KafkaServiceBuilder(SERVICE).editSpec().editTls().withCertificateRef(null).endTls().endSpec().build()));
    }

    @ParameterizedTest
    @MethodSource
    void mappingToSecretToleratesKafkaServicesWithoutTls(KafkaService service) {
        // Given
        EventSourceContext<KafkaService> eventSourceContext = mock();
        KubernetesClient client = mock();
        when(eventSourceContext.getClient()).thenReturn(client);

        KubernetesResourceList<KafkaService> mockList = mockKafkaServiceListOperation(client);
        when(mockList.getItems()).thenReturn(List.of(service));

        // When
        var mapper = KafkaServiceReconciler.secretToKafkaService(eventSourceContext);

        // Then
        var primaryResourceIDs = mapper.toPrimaryResourceIDs(new SecretBuilder().withNewMetadata().withName("secret").endMetadata().build());
        assertThat(primaryResourceIDs).isEmpty();
    }

    private KubernetesResourceList<KafkaService> mockKafkaServiceListOperation(KubernetesClient client) {
        MixedOperation<KafkaService, KubernetesResourceList<KafkaService>, Resource<KafkaService>> mockOperation = mock();
        when(client.resources(KafkaService.class)).thenReturn(mockOperation);
        KubernetesResourceList<KafkaService> mockList = mock();
        when(mockOperation.list()).thenReturn(mockList);
        when(mockOperation.inNamespace(any())).thenReturn(mockOperation);
        return mockList;
    }

    @SuppressWarnings("unchecked")
    private static Context<KafkaService> mockContext() {
        return mock(Context.class);
    }
}
