/*
 * 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.common.requests;

import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.DeleteTopicsRequestData;
import org.apache.kafka.common.message.DeleteTopicsRequestData.DeleteTopicState;

import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

import static org.apache.kafka.common.protocol.ApiKeys.DELETE_TOPICS;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class DeleteTopicsRequestTest {

    @Test
    public void testTopicNormalization() {
        for (short version : DELETE_TOPICS.allVersions()) {
            // Check topic names are in the correct place when using topicNames.
            String topic1 = "topic1";
            String topic2 = "topic2";
            List<String> topics = Arrays.asList(topic1, topic2);
            DeleteTopicsRequest requestWithNames = new DeleteTopicsRequest.Builder(
                    new DeleteTopicsRequestData().setTopicNames(topics)).build(version);
            DeleteTopicsRequest requestWithNamesSerialized = DeleteTopicsRequest.parse(requestWithNames.serialize(), version);

            assertEquals(topics, requestWithNames.topicNames());
            assertEquals(topics, requestWithNamesSerialized.topicNames());

            if (version < 6) {
                assertEquals(topics, requestWithNames.data().topicNames());
                assertEquals(topics, requestWithNamesSerialized.data().topicNames());
            } else {
                // topics in TopicNames are moved to new topics field
                assertEquals(topics, requestWithNames.data().topics().stream().map(DeleteTopicState::name).collect(Collectors.toList()));
                assertEquals(topics, requestWithNamesSerialized.data().topics().stream().map(DeleteTopicState::name).collect(Collectors.toList()));
            }
        }
    }

    @Test
    public void testNewTopicsField() {
        for (short version : DELETE_TOPICS.allVersions()) {
            String topic1 = "topic1";
            String topic2 = "topic2";
            List<String> topics = Arrays.asList(topic1, topic2);
            DeleteTopicsRequest requestWithNames = new DeleteTopicsRequest.Builder(
                    new DeleteTopicsRequestData().setTopics(Arrays.asList(
                            new DeleteTopicsRequestData.DeleteTopicState().setName(topic1),
                            new DeleteTopicsRequestData.DeleteTopicState().setName(topic2)))).build(version);
            // Ensure we only use new topics field on versions 6+.
            if (version >= 6) {
                DeleteTopicsRequest requestWithNamesSerialized = DeleteTopicsRequest.parse(requestWithNames.serialize(), version);

                assertEquals(topics, requestWithNames.topicNames());
                assertEquals(topics, requestWithNamesSerialized.topicNames());

            } else {
                // We should fail if version is less than 6.
                assertThrows(UnsupportedVersionException.class, requestWithNames::serialize);
            }
        }
    }

    @Test
    public void testTopicIdsField() {
        for (short version : DELETE_TOPICS.allVersions()) {
            // Check topic IDs are handled correctly. We should only use this field on versions 6+.
            Uuid topicId1 = Uuid.randomUuid();
            Uuid topicId2 = Uuid.randomUuid();
            List<Uuid> topicIds = Arrays.asList(topicId1, topicId2);
            DeleteTopicsRequest requestWithIds = new DeleteTopicsRequest.Builder(
                    new DeleteTopicsRequestData().setTopics(Arrays.asList(
                            new DeleteTopicsRequestData.DeleteTopicState().setTopicId(topicId1),
                            new DeleteTopicsRequestData.DeleteTopicState().setTopicId(topicId2)))).build(version);

            if (version >= 6) {
                DeleteTopicsRequest requestWithIdsSerialized = DeleteTopicsRequest.parse(requestWithIds.serialize(), version);

                assertEquals(topicIds, requestWithIds.topicIds());
                assertEquals(topicIds, requestWithIdsSerialized.topicIds());

                // All topic names should be replaced with null
                requestWithIds.data().topics().forEach(topic -> assertNull(topic.name()));
                requestWithIdsSerialized.data().topics().forEach(topic -> assertNull(topic.name()));
            } else {
                // We should fail if version is less than 6.
                assertThrows(UnsupportedVersionException.class, requestWithIds::serialize);
            }
        }
    }

    @Test
    public void testDeleteTopicsRequestNumTopics() {
        for (short version : DELETE_TOPICS.allVersions()) {
            DeleteTopicsRequest request = new DeleteTopicsRequest.Builder(new DeleteTopicsRequestData()
                    .setTopicNames(Arrays.asList("topic1", "topic2"))
                    .setTimeoutMs(1000)).build(version);
            DeleteTopicsRequest serializedRequest = DeleteTopicsRequest.parse(request.serialize(), version);
            // createDeleteTopicsRequest sets 2 topics
            assertEquals(2, request.numberOfTopics());
            assertEquals(2, serializedRequest.numberOfTopics());

            // Test using IDs
            if (version >= 6) {
                DeleteTopicsRequest requestWithIds = new DeleteTopicsRequest.Builder(
                        new DeleteTopicsRequestData().setTopics(Arrays.asList(
                                new DeleteTopicsRequestData.DeleteTopicState().setTopicId(Uuid.randomUuid()),
                                new DeleteTopicsRequestData.DeleteTopicState().setTopicId(Uuid.randomUuid())))).build(version);
                DeleteTopicsRequest serializedRequestWithIds = DeleteTopicsRequest.parse(requestWithIds.serialize(), version);
                assertEquals(2, requestWithIds.numberOfTopics());
                assertEquals(2, serializedRequestWithIds.numberOfTopics());
            }
        }
    }
}
