/*
 * 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.streams.tests;

import org.apache.kafka.common.serialization.IntegerDeserializer;
import org.apache.kafka.common.serialization.IntegerSerializer;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.TopologyTestDriver;
import org.apache.kafka.test.TestUtils;

import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;

import java.util.Map;
import java.util.TreeMap;

import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;

@Timeout(600)
public class RelationalSmokeTestTest extends SmokeTestUtil {

    @Test
    public void verifySmokeTestLogic() {
        try (final TopologyTestDriver driver =
                 new TopologyTestDriver(RelationalSmokeTest.App.getTopology(),
                                        RelationalSmokeTest.App.getConfig(
                                            "nothing:0",
                                            "test",
                                            "test",
                                            StreamsConfig.AT_LEAST_ONCE,
                                            "classic",
                                            TestUtils.tempDirectory().getAbsolutePath()
                                        ))) {

            final TestInputTopic<Integer, RelationalSmokeTest.Article> articles =
                driver.createInputTopic(RelationalSmokeTest.ARTICLE_SOURCE,
                                        new IntegerSerializer(),
                                        new RelationalSmokeTest.Article.ArticleSerializer());

            final TestInputTopic<Integer, RelationalSmokeTest.Comment> comments =
                driver.createInputTopic(RelationalSmokeTest.COMMENT_SOURCE,
                                        new IntegerSerializer(),
                                        new RelationalSmokeTest.Comment.CommentSerializer());

            final TestOutputTopic<Integer, RelationalSmokeTest.AugmentedArticle> augmentedArticles =
                driver.createOutputTopic(RelationalSmokeTest.ARTICLE_RESULT_SINK,
                                         new IntegerDeserializer(),
                                         new RelationalSmokeTest.AugmentedArticle.AugmentedArticleDeserializer());

            final TestOutputTopic<Integer, RelationalSmokeTest.AugmentedComment> augmentedComments =
                driver.createOutputTopic(RelationalSmokeTest.COMMENT_RESULT_SINK,
                                         new IntegerDeserializer(),
                                         new RelationalSmokeTest.AugmentedComment.AugmentedCommentDeserializer());

            final RelationalSmokeTest.DataSet dataSet =
                RelationalSmokeTest.DataSet.generate(10, 30);

            final Map<Integer, RelationalSmokeTest.Article> articleMap = new TreeMap<>();
            for (final RelationalSmokeTest.Article article : dataSet.getArticles()) {
                articles.pipeInput(article.getKey(), article, article.getTimestamp());
                articleMap.put(article.getKey(), article);
            }

            final Map<Integer, Long> commentCounts = new TreeMap<>();

            final Map<Integer, RelationalSmokeTest.Comment> commentMap = new TreeMap<>();
            for (final RelationalSmokeTest.Comment comment : dataSet.getComments()) {
                comments.pipeInput(comment.getKey(), comment, comment.getTimestamp());
                commentMap.put(comment.getKey(), comment);
                commentCounts.put(comment.getArticleId(),
                                  commentCounts.getOrDefault(comment.getArticleId(), 0L) + 1);
            }

            final Map<Integer, RelationalSmokeTest.AugmentedArticle> augmentedArticleResults =
                augmentedArticles.readKeyValuesToMap();

            final Map<Integer, RelationalSmokeTest.AugmentedComment> augmentedCommentResults =
                augmentedComments.readKeyValuesToMap();

            assertThat(augmentedArticleResults.size(), is(dataSet.getArticles().length));
            assertThat(augmentedCommentResults.size(), is(dataSet.getComments().length));

            assertThat(
                RelationalSmokeTest.App.verifySync(true,
                                                   articleMap,
                                                   commentMap,
                                                   augmentedArticleResults,
                                                   augmentedCommentResults),
                is(true));
        }
    }
}