package org.apache.kafka.streams.test;

public class TestInputTopic<K, V>TopologyTestDriver {
    public TopologyTestDriver(Topology topology, Properties config); //Timestamp handlingexisting constructor
    //Record timestamp can be provided when piping input or use internally tracked time initialized with constructors:
    //startTimestampMs the initial timestamp for generated records, if not provided uses current system time as start timestamp.
    //autoAdvanceMs the time increment per generated record, if not provided auto-advance is disabled.

    //Constructors with serializers
    TestInputTopic(final TopologyTestDriver driver, @Deprecate public TopologyTestDriver(Topology topology, Properties config, long initialWallClockTimeMs);
    public TopologyTestDriver(Topology topology, Properties config, Instant initialWallClockTime);

    @Deprecate public void advanceWallClockTime(long advanceMs); // can trigger wall-clock-time punctuation
	public void advanceWallClockTime(Duration advance); // can trigger wall-clock-time punctuation

    //Deprecate old pipe and read methods
    @Deprecate public void pipeInput(ConsumerRecord<byte[], byte[]> record); // can trigger event-time punctuation
    @Deprecate public void pipeInput(List<ConsumerRecord<byte[], byte[]>> records); // can trigger event-time punctuation
    @Deprecate public ProducerRecord<byte[], byte[]> readOutput(String topic);
    @Deprecate public <K, V> ProducerRecord<K, V> readOutput(String topic, Deserializer<K> keyDeserializer, Deserializer<V> valueDeserializer);

	// methods for TestTopic object creation
	public final <K, V> TestOutputTopic<K, V> createOutputTopic(final String topicName, final Serializer<K> keySerializer, final Serializer<V> valueSerializer);
	// Uses current system  TestInputTopic(final TopologyTestDriver driver, final String topicName, final Serializer<K> keySerializertime as start timestamp. Auto-advance is disabled.
	public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName, final Serializer<V>Deserializer<K> valueSerializerkeyDeserializer, final longDeserializer<V> startTimestampMsvalueDeserializer);
    //Uses provided  TestInputTopic(final TopologyTestDriver driver, startTimestamp and autoAdvance duration for timestamp generation
	public final <K, V> TestInputTopic<K, V> createInputTopic(final String topicName, final Serializer<K>Deserializer<K> keySerializerkeyDeserializer, final Serializer<V>Deserializer<V> valueSerializervalueDeserializer, final longInstant startTimestampMsstartTimestamp, final longDuration autoAdvance);

    ...
}

package org.apache.kafka.streams.test;

public class TestOutputTopic<KTestRecord<K, V> {
	//Constructor with serializers
Constructors
    public TestRecord(final V value);
    public TestRecord(final K key, final V value);
    public TestOutputTopic(TestRecord(final K key, final TopologyTestDriverV drivervalue, final String topic Headers headers);
    public TestRecord(final K key, final Serde<K>V keySerdevalue, final Serde<V>Instant valueSerderecordTime);
    public TestRecord(final K key, final V value, final Headers headers, //Constructorfinal withInstant serdesrecordTime);
    TestOutputTopicpublic TestRecord(final TopologyTestDriverK driverkey, final StringV topicvalue, final Deserializer<K>Headers keyDeserializerheaders, final Deserializer<V>Long valueDeserializertimestamp);
	
	//Methods to readOutputConstructor based on existing record
    public TestRecord(final ConsumerRecord<K, V> record);
    public TestRecord(final ProducerRecord<K, V> readRecord record);
	
	// Methods like in ProducerRecord / ConsumerRecord
    public Headers headers();
    KeyValue<K,public V>K readKeyValuekey();
    public V readValuevalue();

    public Long //Output as collectiontimestamp();

	// Getters
    public Headers getHeaders();
    public K getKey();
    public V getValue();
    public Instant getRecordTime();
	
	//Overrides
    Map<K,public V>String readKeyValuesToMaptoString();
    List<KeyValue<K,public V>>boolean readKeyValuesToListequals(Object o);
    public List<V>int readValuesToListhashCode();
}

public class SimpleTopicTest {

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

  @Before
  public void setup() {
    testDriver = new TopologyTestDriver(TestStream.getTopology(), TestStream.getConfig());
    	inputTopic = new TestInputTopic<>(testDriver, testDriver.createInputTopic(TestStream.INPUT_TOPIC, new Serdes.StringSerdeStringDeserializer(), new Serdes.StringSerdeStringDeserializer());
    outputTopic = new TestOutputTopic<>(testDriver, testDriver.createOutputTopic(TestStream.OUTPUT_TOPIC, new Serdes.StringSerdeStringSerializer(), new Serdes.StringSerdeLongSerializer());
  }

  @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.readRecordisEmpty()).isNullisTrue();
  }
}