package org.apache.kafka.streams.test;

public class TestOutputTopic<K, V> {
	//Constructor with serializers
    TestOutputTopic(final TopologyTestDriver driver, final String topic, final Serde<K> keySerde, final Serde<V> valueSerde);
    //Constructor with serdes
    TestOutputTopic(final TopologyTestDriver driver, final String topic, final Deserializer<K> keyDeserializer, final Deserializer<V> valueDeserializer);

	//Methods to readOutput
    ProducerRecord<K, V> readRecord();
    KeyValue<K, V> readKeyValue();
    V readValue();

    //Output as collection
    Map<K, V> readKeyValuesToMap();
    List<KeyValue<K, V>> readKeyValuesToList();
    List<V> readValuesToList();

	//Possible If KPI-451 is included
    Iterable<ProducerRecord<K, V>> iterableRecords();
    Iterable<KeyValue<K, V>> iterableKeyValues();
    Iterable<V> iterableValues();
}

public class SimpleTopicTest {

  private TopologyTestDriver testDriver;
  private TestInputTopic<String, String> inputTopic;
  private TestOutputTopic<String, String> outputTopic;

  @Before
  public void setup() {
    final StreamsBuilder buildertestDriver = new StreamsBuilder();
    TestStream app = new TestStream();
    //Create Actual Stream Processing pipeline
    app.createStream(builder);
    testDriver = new TopologyTestDriver(builder.build(),app.configTopologyTestDriver(TestStream.getTopology(), TestStream.getConfig());
    inputTopic = new TestInputTopic<String, String>(testDriver, TestStream.INPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
    outputTopic = new TestOutputTopic<String, String>(testDriver, TestStream.OUTPUT_TOPIC, new Serdes.StringSerde(), new Serdes.StringSerde());
  }

  @After
  public void tearDown() {
      testDriver.close();
  }

  @Test
  public void testOneWord() {
    //Feed word "Hello" to inputTopic and no kafka key, timestamp is irrelevant in this case
    inputTopic.pipeInput("Hello");
    assertThat(outputTopic.readValue()).isEqualTo("Hello");
    //No more output in topic
    assertThat(outputTopic.readRecord()).isNull();
  }
}