/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements. See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You 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 org.apache.kafka.tools;

import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.common.test.ClusterInstance;
import org.apache.kafka.common.test.api.ClusterTest;
import org.apache.kafka.common.test.api.ClusterTests;
import org.apache.kafka.common.test.api.Type;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.api.Test;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;

class MetadataQuorumCommandTest {

    /**
     * 1. The same number of broker controllers
     * 2. More brokers than controllers
     * 3. Fewer brokers than controllers
     */
    @ClusterTests({
        @ClusterTest(brokers = 2, controllers = 2),
        @ClusterTest(brokers = 2, controllers = 1),
        @ClusterTest(brokers = 1, controllers = 2),
    })
    public void testDescribeQuorumReplicationSuccessful(ClusterInstance cluster) throws InterruptedException {
        cluster.waitForReadyBrokers();
        String describeOutput = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--replication")
        );

        List<String> outputs = Arrays.stream(describeOutput.split("\n")).toList();
        String header = outputs.get(0);
        List<String> data = outputs.subList(1, outputs.size());

        assertTrue(header.matches("NodeId\\s+DirectoryId\\s+LogEndOffset\\s+Lag\\s+LastFetchTimestamp\\s+LastCaughtUpTimestamp\\s+Status\\s+"));

        if (cluster.type() == Type.CO_KRAFT) 
            assertEquals(Math.max(cluster.config().numControllers(), cluster.config().numBrokers()), data.size());
        else 
            assertEquals(cluster.config().numBrokers() + cluster.config().numControllers(), data.size());

        Pattern leaderPattern = Pattern.compile("\\d+\\s+\\S+\\s+\\d+\\s+\\d+\\s+-?\\d+\\s+-?\\d+\\s+Leader\\s*");
        assertTrue(leaderPattern.matcher(data.get(0)).find());
        assertTrue(data.stream().skip(1).noneMatch(o -> leaderPattern.matcher(o).find()));

        Pattern followerPattern = Pattern.compile("\\d+\\s+\\S+\\s+\\d+\\s+\\d+\\s+-?\\d+\\s+-?\\d+\\s+Follower\\s*");
        assertEquals(cluster.config().numControllers() - 1, data.stream().filter(o -> followerPattern.matcher(o).find()).count());

        Pattern observerPattern = Pattern.compile("\\d+\\s+\\S+\\s+\\d+\\s+\\d+\\s+-?\\d+\\s+-?\\d+\\s+Observer\\s*");
        if (cluster.type() == Type.CO_KRAFT)
            assertEquals(Math.max(0, cluster.config().numBrokers() - cluster.config().numControllers()),
                data.stream().filter(o -> observerPattern.matcher(o).find()).count());
        else
            assertEquals(cluster.config().numBrokers(), data.stream().filter(o -> observerPattern.matcher(o).find()).count());
    }

    /**
     * 1. The same number of broker controllers
     * 2. More brokers than controllers
     * 3. Fewer brokers than controllers
     */
    @ClusterTests({
        @ClusterTest(brokers = 2, controllers = 2),
        @ClusterTest(brokers = 2, controllers = 1),
        @ClusterTest(brokers = 1, controllers = 2),
    })
    public void testDescribeQuorumStatusSuccessful(ClusterInstance cluster) throws InterruptedException {
        testDescribeQuorumStatusSuccessful(cluster, false);
        testDescribeQuorumStatusSuccessful(cluster, true);
    }

    private void testDescribeQuorumStatusSuccessful(ClusterInstance cluster, boolean usingBootstrapController) throws InterruptedException {
        cluster.waitForReadyBrokers();

        String describeOutput = ToolsTestUtils.captureStandardOut(
            () -> MetadataQuorumCommand.mainNoExit(
                usingBootstrapController ? "--bootstrap-controller" : "--bootstrap-server",
                usingBootstrapController ? cluster.bootstrapControllers() : cluster.bootstrapServers(),
                "describe",
                "--status"
            )
        );
        String[] outputs = describeOutput.split("\n");

        assertTrue(outputs[0].matches("ClusterId:\\s+\\S{22}"), describeOutput);
        assertTrue(outputs[1].matches("LeaderId:\\s+\\d+"), describeOutput);
        assertTrue(outputs[2].matches("LeaderEpoch:\\s+\\d+"), describeOutput);
        // HighWatermark may be -1
        assertTrue(outputs[3].matches("HighWatermark:\\s+-?\\d+"), describeOutput);
        assertTrue(outputs[4].matches("MaxFollowerLag:\\s+\\d+"), describeOutput);
        assertTrue(outputs[5].matches("MaxFollowerLagTimeMs:\\s+-?\\d+"), describeOutput);
        assertTrue(
            outputs[6].matches("CurrentVoters:\\s+\\[\\{\"id\":\\s+\\d+,\\s+" +
                "\"endpoints\":\\s+\\[\"\\S+://\\[?\\S+]?:\\d+\",?.*]"),
            describeOutput
        );

        // There are no observers if we have fewer brokers than controllers
        if (cluster.type() == Type.CO_KRAFT && cluster.config().numBrokers() <= cluster.config().numControllers()) {
            assertTrue(outputs[7].matches("CurrentObservers:\\s+\\[]"), describeOutput);
        } else {
            assertTrue(
                outputs[7].matches("CurrentObservers:\\s+\\[\\{\"id\":\\s+\\d+,\\s+\"directoryId\":\\s+\\S+}" +
                    "(,\\s+\\{\"id\":\\s+\\d+,\\s+\"directoryId\":\\s+\\S+})*]"),
                describeOutput
            );
        }
    }

    @ClusterTest
    public void testOnlyOneBrokerAndOneController(ClusterInstance cluster) {
        testOnlyOneBrokerAndOneController(cluster, false);
        testOnlyOneBrokerAndOneController(cluster, true);
    }

    public void testOnlyOneBrokerAndOneController(ClusterInstance cluster, boolean usingBootstrapController) {
        String statusOutput = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit(
                    usingBootstrapController ? "--bootstrap-controller" : "--bootstrap-server",
                    usingBootstrapController ? cluster.bootstrapControllers() : cluster.bootstrapServers(),
                    "describe", "--status")
        );
        assertEquals("MaxFollowerLag:         0", statusOutput.split("\n")[4]);
        assertEquals("MaxFollowerLagTimeMs:   0", statusOutput.split("\n")[5]);

        String replicationOutput = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit(
                    usingBootstrapController ? "--bootstrap-controller" : "--bootstrap-server",
                    usingBootstrapController ? cluster.bootstrapControllers() : cluster.bootstrapServers(),
                    "describe", "--replication")
        );
        assertEquals("0", replicationOutput.split("\n")[1].split("\\s+")[3]);
    }

    @Test
    public void testCommandConfig() throws IOException {
        // specifying a --command-config containing properties that would prevent login must fail
        File tmpfile = TestUtils.tempFile(AdminClientConfig.SECURITY_PROTOCOL_CONFIG + "=SSL_PLAINTEXT");
        assertEquals(1, MetadataQuorumCommand.mainNoExit("--bootstrap-server", "localhost:9092",
                        "--command-config", tmpfile.getAbsolutePath(), "describe", "--status"));
    }

    @ClusterTest(types = {Type.CO_KRAFT})
    public void testHumanReadableOutput(ClusterInstance cluster) {
        assertEquals(1, MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--human-readable"));
        assertEquals(1, MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--status", "--human-readable"));
        String out0 = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--replication")
        );
        assertFalse(out0.split("\n")[1].matches("\\d*"));
        String out1 = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--replication", "--human-readable")
        );
        assertHumanReadable(out1);
        String out2 = ToolsTestUtils.captureStandardOut(() ->
            MetadataQuorumCommand.mainNoExit("--bootstrap-server", cluster.bootstrapServers(), "describe", "--re", "--hu")
         );
        assertHumanReadable(out2);
    }

    private static void assertHumanReadable(String output) {
        String dataRow = output.split("\n")[1];
        String lastFetchTimestamp = dataRow.split("\t")[4];
        String lastFetchTimestampValue = lastFetchTimestamp.split(" ")[0];
        String lastCaughtUpTimestamp = dataRow.split("\t")[5];
        String lastCaughtUpTimestampValue = lastCaughtUpTimestamp.split(" ")[0];
        assertTrue(lastFetchTimestamp.contains("ms ago"));
        assertTrue(lastFetchTimestampValue.matches("\\d*"));
        assertTrue(lastCaughtUpTimestamp.contains("ms ago"));
        assertTrue(lastCaughtUpTimestampValue.matches("\\d*"));
    }
}
