/*
 * Copyright Strimzi authors.
 * License: Apache License 2.0 (see the file LICENSE or http://apache.org/licenses/LICENSE-2.0.html).
 */
package io.strimzi.operator.cluster.operator.assembly;

import java.util.ArrayList;
import java.util.Base64;
import java.util.List;
import java.util.function.Function;

import io.fabric8.kubernetes.api.model.Pod;
import io.fabric8.kubernetes.api.model.PodBuilder;
import io.fabric8.kubernetes.api.model.Secret;
import io.fabric8.kubernetes.api.model.SecretBuilder;
import io.fabric8.kubernetes.api.model.apps.StatefulSet;
import io.fabric8.kubernetes.client.KubernetesClient;
import io.strimzi.api.kafka.Crds;
import io.strimzi.api.kafka.KafkaList;
import io.strimzi.api.kafka.model.Kafka;
import io.strimzi.api.kafka.model.KafkaBuilder;
import io.strimzi.api.kafka.model.KafkaResources;
import io.strimzi.api.kafka.model.listener.arraylistener.KafkaListenerType;
import io.strimzi.certs.CertManager;
import io.strimzi.operator.KubernetesVersion;
import io.strimzi.operator.PlatformFeaturesAvailability;
import io.strimzi.operator.cluster.ClusterOperatorConfig;
import io.strimzi.operator.cluster.KafkaVersionTestUtils;
import io.strimzi.operator.cluster.ResourceUtils;
import io.strimzi.operator.cluster.model.KafkaCluster;
import io.strimzi.operator.cluster.model.KafkaVersion;
import io.strimzi.operator.cluster.operator.resource.ResourceOperatorSupplier;
import io.strimzi.operator.cluster.operator.resource.ZookeeperLeaderFinder;
import io.strimzi.operator.cluster.operator.resource.ZookeeperScalerProvider;
import io.strimzi.operator.common.AdminClientProvider;
import io.strimzi.operator.common.MetricsProvider;
import io.strimzi.operator.common.PasswordGenerator;
import io.strimzi.operator.common.Reconciliation;
import io.strimzi.operator.common.operator.MockCertManager;
import io.strimzi.test.mockkube.MockKube;
import io.vertx.core.Future;
import io.vertx.core.Vertx;
import io.vertx.junit5.Checkpoint;
import io.vertx.junit5.VertxExtension;
import io.vertx.junit5.VertxTestContext;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasEntry;
import static org.hamcrest.Matchers.hasKey;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.not;
import static org.mockito.Mockito.mock;

@ExtendWith(VertxExtension.class)
public class KafkaAssemblyOperatorCustomCertTest {
    private final KubernetesVersion kubernetesVersion = KubernetesVersion.V1_18;
    private final MockCertManager certManager = new MockCertManager();
    private final PasswordGenerator passwordGenerator = new PasswordGenerator(10, "a", "a");
    private final ClusterOperatorConfig config = ResourceUtils.dummyClusterOperatorConfig(VERSIONS);
    private static final KafkaVersion.Lookup VERSIONS = KafkaVersionTestUtils.getKafkaVersionLookup();
    private final String namespace = "testns";
    private final String clusterName = "testkafka";
    protected static Vertx vertx;

    private Kafka kafka;
    private KafkaAssemblyOperator operator;
    private KubernetesClient client;
    private final List<Function<Pod, List<String>>> functionArgumentCaptor = new ArrayList<>();

    /**
     * Mock the KafkaAssemblyOperator and override reconcile to only run through the steps we want to test
     */
    class MockKafkaAssemblyOperator extends KafkaAssemblyOperator  {

        public MockKafkaAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa, CertManager certManager, PasswordGenerator passwordGenerator, ResourceOperatorSupplier supplier, ClusterOperatorConfig config) {
            super(vertx, pfa, certManager, passwordGenerator, supplier, config);
        }

        @Override
        ReconciliationState createReconciliationState(Reconciliation reconciliation, Kafka kafkaAssembly) {
            return new ReconciliationState(reconciliation, kafkaAssembly) {
                @Override
                Future<Void> maybeRollKafka(StatefulSet sts, Function<Pod, List<String>> podNeedsRestart) {
                    functionArgumentCaptor.add(podNeedsRestart);
                    return Future.succeededFuture();
                }
            };
        }

        @Override
        Future<Void> reconcile(ReconciliationState reconcileState)  {
            return reconcileState.reconcileCas(this::dateSupplier)
                    .compose(state -> state.getKafkaClusterDescription())
                    .compose(state -> state.customListenerCertificates())
                    .compose(state -> state.kafkaStatefulSet())
                    .compose(state -> state.kafkaRollingUpdate())
                    .map((Void) null);
        }
    }

    @BeforeAll
    public static void before() {
        vertx = Vertx.vertx();
    }

    @BeforeEach
    public void setup() {
        kafka = createKafka();

        client = new MockKube()
                .withCustomResourceDefinition(Crds.kafka(), Kafka.class, KafkaList.class).end()
                .build();
        Crds.kafkaOperation(client).inNamespace(namespace).create(kafka);
        client.secrets().inNamespace(namespace).create(getTlsSecret());
        client.secrets().inNamespace(namespace).create(getExternalSecret());
        Secret secret = new SecretBuilder()
            .withNewMetadata()
                .withNamespace(namespace)
                .withName("testkafka-cluster-operator-certs")
            .endMetadata()
            .addToData("foo", "bar")
            .build();
        client.secrets().inNamespace(namespace).create(secret);
        ResourceOperatorSupplier supplier = new ResourceOperatorSupplier(vertx, client, mock(ZookeeperLeaderFinder.class),
                mock(AdminClientProvider.class), mock(ZookeeperScalerProvider.class),
                mock(MetricsProvider.class), new PlatformFeaturesAvailability(false, KubernetesVersion.V1_20), 10000);
        operator = new MockKafkaAssemblyOperator(vertx, new PlatformFeaturesAvailability(false, kubernetesVersion),
                certManager,
                passwordGenerator,
                supplier,
                config);

    }

    @AfterAll
    public static void after() {
        vertx.close();
    }

    public Kafka createKafka() {
        return new KafkaBuilder()
                .withNewMetadata()
                    .withName(clusterName)
                    .withNamespace(namespace)
                    .withGeneration(2L)
                .endMetadata()
                .withNewSpec()
                    .withNewKafka()
                        .withReplicas(3)
                        .withNewListeners()
                            .addNewGenericKafkaListener()
                                .withName("tls")
                                .withPort(9093)
                                .withType(KafkaListenerType.INTERNAL)
                                .withTls(true)
                                .withNewConfiguration()
                                    .withNewBrokerCertChainAndKey()
                                        .withSecretName("my-tls-secret")
                                        .withCertificate("tls.crt")
                                        .withKey("tls.key")
                                    .endBrokerCertChainAndKey()
                                .endConfiguration()
                            .endGenericKafkaListener()
                            .addNewGenericKafkaListener()
                                .withName("external")
                                .withPort(9094)
                                .withType(KafkaListenerType.NODEPORT)
                                .withTls(true)
                                .withNewConfiguration()
                                    .withNewBrokerCertChainAndKey()
                                        .withSecretName("my-external-secret")
                                        .withCertificate("tls.crt")
                                        .withKey("tls.key")
                                    .endBrokerCertChainAndKey()
                                .endConfiguration()
                            .endGenericKafkaListener()
                        .endListeners()
                        .withNewEphemeralStorage()
                        .endEphemeralStorage()
                    .endKafka()
                    .withNewZookeeper()
                        .withReplicas(3)
                        .withNewEphemeralStorage()
                        .endEphemeralStorage()
                    .endZookeeper()
                .endSpec()
                .build();
    }

    public Secret getTlsSecret() {
        return new SecretBuilder()
                .withNewMetadata()
                    .withNewName("my-tls-secret")
                .endMetadata()
                .addToData("tls.crt", " ")
                .addToData("tls.key", "LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0tCk1JSUV2Z0lCQURBTkJna3Foa2lHOXcwQkFRRUZBQVNDQktnd2dnU2tBZ0VBQW9JQkFRRGNIWEt5c0ZXNnZlaU8KOVY1YTJMY2VURGNCOHllbldYTHZkNFNGWGR0MGVPSU1WZ0FHYUlYVXNseFdwKzlQVFI2RkpaQzcxZW8zVVBoeApEcU0xcWw3V1VuYlFzVmxseTlZc2dyeVAxNk4yd3hSUTdxUnNvTG1VcE5EWTdqTTlObE9ic01UMGgxSnIzRmhYCk9CbC9meVliVWlaZ2QvbVhjVjExYU5DKzhZK0M1U3pFTWVreGJIRWxrUkNkUHpIWVFqajNBMGg2Z3pLWS9lNEkKQlpuZEVrNkplejF5ZmNNL0cvL3JjdFFhUjEwdTlwcUVXcG85TkJZQXVnYU1KRmZudUJxRDZRckhDdWxjQm1mNgpUNmpZcGpHYS9KL0M2dTZGV3BQTitVbWV5eC94T0FlVnpMZ1pVemU2cWZRU3pyNTRReDBJWTYra1FiaEVGTVRDCkFJWjRUNFhCQWdNQkFBRUNnZ0VCQU5JblAzb0JSMmlLcG4zUElLM2wyVGVSRnJmQzJxbzVpYlcrUzRVMXJqQU8KdGV1SE5IRHAzRlROa2NHZWhxb1UvRDJ0TnZsUGJGWXg5WEdqd3dtYXh2OGpMcE5qci9HejRxRU9sVTlVVjVvcworTG1vanMyenlsdHozSDR4TmpTTUtOa3R0VzJ0d1hCL3FNeGxJRnNOSDJuWVRoR3VtbHNQL21YNWs4dXFRQlY4ClNMQk13U1RodXlaZDgydEw1YTlhYTV0cHVpRkNpbm13QVF3OGRpMVZGaWwzbVdaZHpySnkxeWNzVWxYeUg0K2sKcVdRUVpUZFJ0bGVDTENTZi85VkFIMmt1SXVvcE1yZWxxeVArNWlDUlp0L2J1VUltNVV4RWNxbzVOcVJ3TlFsWQp2eXAwWjIxaXgzMEliaXNMdlc3cWk5ZE1LZzdHamhBdlBFWEhiSENHWnkwQ2dZRUEzaUFwOVUxbWFmcTZIcmZqCjNieGJNMCttTHBYaDBUYXk3bXhoK0tpeGtwbGNzeW5qQmFHZnhpUVpaSkZGMG5qaG51Q2ltaCs5SE4velFCN1oKbjVGZ2QxS2JjQjM2WGdObUJUQ01CMmFnb2ZqZXd5enJ3V2ZkQStVc0lzdk12aVBKbUZGeVJFUURUTGVZRnFLNwozeEJ6a05INVRDREtQTzh5TW8wQ1BnejZIZThDZ1lFQS9hN0tIVHN4eEwxZ0JiOEQ4UUhhTkR3SG4xVWsyRVlNClk3d1J1ZTcrNGlIdnc2Z2kvWnBXT1FhZHQrQ2JzWml2UjYydGovQ0hteHVjYjJSMVE2Qlh3ZXZuUkVwdVlsS2wKd0NJN09Hb1ZFK2ZXeWxHTVBvVG1nNGZCUjNNM09yeG9mT0RMenRhSzl3VllTL3JtNEZ3Mk9yZXhzUHpWalFEUgpuVnduSWxkZFIwOENnWUVBMVlmRDdnMW02M0JjNVZUUGw0UVBoQ2NJVVBaQ3E5VlNjMEw3ZDRmcmxFc2J3eHY4CklwaTV1RWRScGN6RmUwdzdVSGtQdHV2VjUyRWVQVUNxNGV0bCthOE92OXdCcDhqS2xTaVRKRFl6S3lITU80SCsKYk9GRXBRNzB1OHFBMnpRYUF2UWd6YUU0THRLN1FOOVZqVjBLUzJpZXArRkpxUVFrbEZYYmx2endvRDhDZ1lCcApiZ3czeTlNcVJkNHpaU2laTUVEa2RwSmdhTDF3V09SclNzMC9MaEdtSDU2Sy9VVFZpeUFNZ1RCcExDTG8wMkQ5CmREUHUzM01zUm5Sa1l5Yk5IVVY3cGJRdTBKUkJyc0dPTVd2VlRWbEhOWkl4OFdSTTAyVU9Bd3lUeWxHSXlxYk8KUjRyTWdxT3NkLzh6VEtwSlVtbURTN2JBck1OLzMzZytZdjhzcVl4dHh3S0JnRVYzdWprbjJQSmtPMEVlWTBZNQpYSFR2UFVzS3BNMlZRend1aXBmemFrT2pCYkg4bWlmUkJFOFR4TDZCVFlSc3crbGFYVXcxbEQ4aWJDNmRuZFRvCldWRWM3Z2kzRlh1RjBUV3p2K2dIbnFCZWFwbWdxOGE4TEhqVHRFMDNnVmo0aG1vZHJIbkVEZ3J5ajlsc0sxck0KYnp6RTlkcWdrNU5CL0c4QmwvTlJnd01BCi0tLS0tRU5EIFBSSVZBVEUgS0VZLS0tLS0K")
                .build();
    }

    public String getTlsThumbprint()    {
        return "{external=COWn2zWLZMhoewfrmSTfUeKlQPifBKekyXzjm2iGTuc=, tls=vjPd/D/f0/X3yqitf65yoUZbyeWnQU4cPDJGbr7GA7I=}";
    }

    public Secret getExternalSecret() {
        return new SecretBuilder()
                .withNewMetadata()
                    .withNewName("my-external-secret")
                .endMetadata()
                .addToData("tls.crt", " ")
                .addToData("tls.key", "LS0tLS1CRUdJTiBQUklWQVRFIEtFWS0tLS0tCk1JSUV2Z0lCQURBTkJna3Foa2lHOXcwQkFRRUZBQVNDQktnd2dnU2tBZ0VBQW9JQkFRQ1ZGQytndW9XNjdmWEQKNmQvNWNhcjhzM3JLTHhUY0pYM09GeGJJdytzVDZ2Szh4NXFQUlY5Q0toeWRyc1hlYTZ0UFg3YUVCRE1UL1lGdwpPMUFnQUtLM01MMEFxU2RWekZLQUJyMnZ4d1NTOURxSklvc29aL0tmZHRmdHRwdUp0QnFmd3lndEI2MVlxU24xClZ3bk1BU2w3QlB5bnNmV01uNEZKakJYOHgwSkNZSGxoQzl1bHM5OWxUUmVZUTYzR1IyTVNKamxVaWJodU83RjcKWWtjZm02SjJEaytEc1V3Yk9zaDgrR1BQRitmam5FOWgyZEQ1SlFXcVI3NGNnajZ1THRNa2dZallNdS9leTZYTgpCWVJBd3NYTnlPbC9VZ0VtVzlQZm95WEUzUVZ0UmFRUGpoOTd0WDY5TWFETUl2TC9mRXlPTXJYRllDU2NlN0szCjBKRW1vbnIvQWdNQkFBRUNnZ0VBTzVhNkF2RUxpMUNhc0JqSDRobEJVNGthUjc3U0E3MG9zRHdpYTFXRW5ZMkkKUVZVM3ZwVG9JclphZ2R6ZVVxMk82RWRGMlRja2c1VU5MQ05KUDhHQlNPQStiQWt4SStac0E2aXVJWmpYaHpZQQpQOWlDN3orOWgyZ2xuMnNpZU1SNDcrcytIK0cxdEg3SnVydHp1d3VyM1BSOVdUcVZBQVN4MVFnZHNkQ2o5NHVrClJzdGsrSGhjM2thT1U0UHM1cWFWbWJZdFB6ZlVibjhoK0xZOWNpZGxQanhiaWZCMTdiK0FaSW9mS2FTb1hEVG4KRks4Wmk1V056cklFdm1TQVZJT00vMHhPbUpnTXdIWUdvUW9PbWJ1TGl0VEFOemQwTmFUVFcyTDV0dlZYa1psOApGZlRrOFRGZ2p2RUpSdExEdklHd3VlS1I1ZE51cGpWSUQ1ek55c0NLSVFLQmdRREFRWVhQclVtd0xheXBhQU5aCmNuY21sL1NNbmowRklwcTJId3Ayb05JRTU1ZlBzZ04yK2tpQklPNXVxYkFGa2U5aGJldTVqS3FLT1hRVFFya3UKa04ybmZvRGJNYklVS29vZXowZyt3REdHeWhJOThzS1REdTZmZTd2Z1FSR3d5ZzIvMUhlRlZVbFMzSHRGbm9aWQpMbnBubXUwcmxoSXg2R2MzaE9PZ1VQN2Jvd0tCZ1FER2dkYUpCVGtuRDVDcjJpYzZ4c2d1WnV6RTd5bzUrZ05uCjREcjc2NXNRc3hNVGhIbjJJMXNpMldBeFl0RmVROGxHMCs3ai9vYjZOUHRaWWg1eXBud2RsbHFiRXVtNUxzYVoKTFZnY09hMGsxeEhxVSt3VUxJWm9hcmlxWXhYUHMvcEU4ZHoralQ3Wld4WmI0aGJ1WmRrU2lJWWVWTzlpaGhKagpKbDZGdTRFWTlRS0JnUUN2S2tPL3J4UUhaK1g3eDEvZDNGUEJId3ZhSHNaYjZtWnBicWk2NHRYWFVDYmFQa2UzCjNGdTVBd2NhWHBLWTBKajQvUXliMXhUK3NWQVh5R0F1bENEUDNZdUxxcUNralFtZy9wekZSNWtZUlA0UDRTSDAKbU5ORERacGt2UVJnUGdmKzhwY2ZMVkNNSllSUEx4c2FOdWFoaE45NEtkaFVEbm9VZEloc1piOSszd0tCZ0d3cApBTGttQkc4WkZ3M2NUdlhDcS81RWpJdjllTGVnVjB5NUs4cHFKTktqa0NoWlRZN2swdHFaTU1XWC8xWnFmdmc5CnIvUEFrdEV3SHlnancwMWJFMU9Yd2dTdStIU3pYUGpIY1RQbjVVU21meGQ3NUsxVldXTDVpMmNqbUJYVkRlK1YKRFlJUmVnWTZrR00rUEpwbkdqRHovSWY0WlhyOGJIWmp5S3I3Y0tzbEFvR0JBTG1TbS8ydkptd21VTzJSQjZEQwpjb1ZrTEZLNkROUmhQaGw4c3NOdDRBWnA4YUJtVzloZC9TditWOHhoNzl6OGhPUlF5cjZoNDVVRjlySXhJb1kyCll3MllidkkyYzRiZlZEQWpGY3U3UEZ3MzVoOFJPRytrMStoNDZQUkhKY1F2dzJxSnN2NnhKOVZLaEdzcEQxdUgKeUVHOEQxRGtVM1FVOElTN01ybUR6Mk9YCi0tLS0tRU5EIFBSSVZBVEUgS0VZLS0tLS0K")
                .build();
    }

    public Pod getPod(StatefulSet sts) {
        return new PodBuilder()
                .withNewMetadataLike(sts.getSpec().getTemplate().getMetadata())
                    .withNewName(KafkaCluster.kafkaClusterName(clusterName) + "-0")
                .endMetadata()
                .withNewSpecLike(sts.getSpec().getTemplate().getSpec())
                .endSpec()
                .build();
    }

    @Test
    public void testPodToRestartIsEmptyWhenCustomCertAnnotationsHaveMatchingThumbprints(VertxTestContext context) {
        Checkpoint async = context.checkpoint();
        operator.createOrUpdate(new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, namespace, clusterName), kafka)
            .onComplete(context.succeeding(v -> context.verify(() -> {
                StatefulSet reconcileSts = client.apps().statefulSets().inNamespace(namespace).withName(KafkaResources.kafkaStatefulSetName(clusterName)).get();
                assertThat(reconcileSts.getSpec().getTemplate().getMetadata().getAnnotations(),
                        hasEntry(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS, getTlsThumbprint()));

                assertThat(functionArgumentCaptor, hasSize(1));
                assertThat(functionArgumentCaptor.get(0).apply(getPod(reconcileSts)), empty());
                async.flag();
            })));
    }

    @Test
    public void testPodToRestartNonemptyWhenCustomCertTlsListenerThumbprintAnnotationsNotMatchingThumbprint(VertxTestContext context) {
        Checkpoint async = context.checkpoint();
        operator.createOrUpdate(new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, namespace, clusterName), kafka)
            .onComplete(context.succeeding(v -> context.verify(() -> {
                StatefulSet reconcileSts = client.apps().statefulSets().inNamespace(namespace).withName(KafkaResources.kafkaStatefulSetName(clusterName)).get();
                assertThat(reconcileSts.getSpec().getTemplate().getMetadata().getAnnotations(),
                        hasEntry(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS, getTlsThumbprint()));

                assertThat(functionArgumentCaptor, hasSize(1));
                Function<Pod, List<String>> isPodToRestart = functionArgumentCaptor.get(0);

                Pod pod = getPod(reconcileSts);
                assertThat("There are no changes in broker config, the restart should not be needed",
                        isPodToRestart.apply(pod), empty());

                pod.getMetadata().getAnnotations().put(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS,
                        Base64.getEncoder().encodeToString("Not the right one!".getBytes()));
                assertThat("Tls listener thumbprint annotation changed, pod should need restart",
                        isPodToRestart.apply(pod).get(0),
                        equalTo("custom certificate one or more listeners changed"));

                async.flag();
            })));
    }

    @Test
    public void testPodToRestartTrueWhenCustomCertExternalListenerThumbprintAnnotationsNotMatchingThumbprint(VertxTestContext context) {
        Checkpoint async = context.checkpoint();
        operator.createOrUpdate(new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, namespace, clusterName), kafka)
            .onComplete(context.succeeding(v -> context.verify(() -> {
                StatefulSet reconcileSts = client.apps().statefulSets().inNamespace(namespace).withName(KafkaResources.kafkaStatefulSetName(clusterName)).get();
                assertThat(reconcileSts.getSpec().getTemplate().getMetadata().getAnnotations(),
                        hasEntry(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS, getTlsThumbprint()));

                assertThat(functionArgumentCaptor, hasSize(1));
                Function<Pod, List<String>> isPodToRestart = functionArgumentCaptor.get(0);

                Pod pod = getPod(reconcileSts);

                assertThat("There are no changes in broker config, the restart should not be needed",
                        isPodToRestart.apply(pod), empty());

                pod.getMetadata().getAnnotations().put(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS,
                        Base64.getEncoder().encodeToString("Not the right one!".getBytes()));

                assertThat(isPodToRestart.apply(pod).get(0),
                        equalTo("custom certificate one or more listeners changed"));

                async.flag();
            })));
    }

    @Test
    public void testPodToRestartIsEmptyAndNoCustomCertAnnotationsWhenNoCustomCertificates(VertxTestContext context) {
        kafka = new KafkaBuilder(createKafka())
                .editSpec()
                    .editKafka()
                        .withNewListeners()
                            .addNewGenericKafkaListener()
                                .withName("tls")
                                .withPort(9093)
                                .withType(KafkaListenerType.INTERNAL)
                                .withTls(true)
                            .endGenericKafkaListener()
                            .addNewGenericKafkaListener()
                                .withName("external")
                                .withPort(9094)
                                .withType(KafkaListenerType.NODEPORT)
                                .withTls(true)
                            .endGenericKafkaListener()
                        .endListeners()
                    .endKafka()
                .endSpec()
                .build();
        Crds.kafkaOperation(client).inNamespace(namespace).withName(clusterName).patch(kafka);


        Checkpoint async = context.checkpoint();
        operator.createOrUpdate(new Reconciliation("test-trigger", Kafka.RESOURCE_KIND, namespace, clusterName), kafka)
            .onComplete(context.succeeding(v -> context.verify(() -> {
                assertThat(functionArgumentCaptor, hasSize(1));

                StatefulSet reconcileSts = client.apps().statefulSets().inNamespace(namespace).withName(KafkaResources.kafkaStatefulSetName(clusterName)).get();
                assertThat(reconcileSts.getSpec().getTemplate().getMetadata().getAnnotations(),
                        not(hasKey(KafkaCluster.ANNO_STRIMZI_CUSTOM_LISTENER_CERT_THUMBPRINTS)));

                List<Function<Pod, List<String>>> capturedFunctions = functionArgumentCaptor;
                assertThat(capturedFunctions, hasSize(1));
                Function<Pod, List<String>> isPodToRestart = capturedFunctions.get(0);
                assertThat(isPodToRestart.apply(getPod(reconcileSts)), empty());

                async.flag();
            })));
    }
}
