package fr.hgwood.todomvckafka;

import com.fasterxml.jackson.databind.ObjectMapper;
import fr.hgwood.todomvckafka.schema.Attribute;
import fr.hgwood.todomvckafka.support.json.JsonSerde;
import fr.hgwood.todomvckafka.support.kafkastreams.TopicInfo;
import fr.hgwood.todomvckafka.support.kafkastreams.Topology;
import fr.hgwood.todomvckafka.support.kafkastreams.TopologyTest;
import io.vavr.jackson.datatype.VavrModule;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.junit.Test;

import static java.util.UUID.randomUUID;
import static org.junit.Assert.assertEquals;

public class FactsToDataTest {
    private static final ObjectMapper OBJECT_MAPPER =
        new ObjectMapper().registerModule(new VavrModule());
    private static final TopicInfo<String, Fact> FACTS = new TopicInfo(
        "test-facts-topic",
        Serdes.String(),
        new JsonSerde(OBJECT_MAPPER, Fact.class)
    );
    private static final TopicInfo<String, TodoItem> TODO_ITEMS = new TopicInfo(
        "test-todo-item-topic",
        Serdes.String(),
        new JsonSerde(OBJECT_MAPPER, TodoItem.class)
    );

    @Test
    public void singleAssertion() throws Exception {
        Topology topology =
            new FactsToDataTopology(FACTS, TODO_ITEMS, OBJECT_MAPPER);

        try (TopologyTest topologyTest = new TopologyTest(topology)) {
            String expectedEntity = "test-entity-id";
            String expectedText = "test-todo-item-text-value";
            KeyValue<String, TodoItem> expected =
                KeyValue.pair(expectedEntity, new TodoItem(expectedText, null));

            KeyValue<String, Fact> input =
                KeyValue.pair(randomUUID().toString(),
                    Fact.of(expectedEntity,
                        Attribute.TODO_ITEM_TEXT,
                        expectedText
                    )
                );
            topologyTest.write(FACTS, input);
            KeyValue<String, TodoItem> actual = topologyTest.read(TODO_ITEMS);

            assertEquals(expected, actual);
        }
    }
}

@@ -2019,24 +2019,1422 @@
 DO_ITEMS);%0A%0A
+            assertEquals(expected, actual);%0A        %7D%0A    %7D%0A%0A    @Test%0A    public void twoAssertion() throws Exception %7B%0A        Topology topology =%0A            new FactsToDataTopology(FACTS, TODO_ITEMS, OBJECT_MAPPER);%0A%0A        try (TopologyTest topologyTest = new TopologyTest(topology)) %7B%0A            String expectedEntity = %22test-entity-id%22;%0A            String expectedText = %22test-todo-item-text-value%22;%0A            Boolean expectedCompletionState = true;%0A            KeyValue%3CString, TodoItem%3E expected =%0A                KeyValue.pair(expectedEntity, new TodoItem(expectedText, expectedCompletionState));%0A%0A            KeyValue%3CString, Fact%3E input1 =%0A                KeyValue.pair(randomUUID().toString(),%0A                    Fact.of(expectedEntity,%0A                        Attribute.TODO_ITEM_TEXT,%0A                        expectedText%0A                    )%0A                );%0A            topologyTest.write(FACTS, input1);%0A            KeyValue%3CString, Fact%3E input2 =%0A                KeyValue.pair(randomUUID().toString(),%0A                    Fact.of(expectedEntity,%0A                        Attribute.TODO_ITEM_COMPLETED,%0A                        expectedCompletionState%0A                    )%0A                );%0A            topologyTest.write(FACTS, input2);%0A            topologyTest.read(TODO_ITEMS); // ignored%0A            KeyValue%3CString, TodoItem%3E actual = topologyTest.read(TODO_ITEMS);%0A%0A
             

FactsToData with two input facts