/**
* TestRunner provides static factory for quick setup of Samza environment for testing Low level api and High level api , users can configure input streams 
* they consume from and output streams they produce to, users pass in their Samza job api logic via StreamTask/AsyncStreamTask/StreamApplication
*/ 

public class TestRunner {
  
  // Static Factory to config & create runner for low level api 
  public static TestRunner of(Class taskClass) {...}
  // Static Factory to config & create runner for high level api
  public static TestRunner of(StreamApplication app) {...}
  
  // Add/Ovveride any custom configs
  public TestRunner addConfigs(Map<String,String> configs) {...}
  public TestRunner addConfigs(Config configs) {...}
  public TestRunner addOverrideConfig(String key, String val){...}
 
  // Configure state for application
  public TestRunner addState(String storeName) {...}  
  
  // Configure an input stream for samza system, that job can consume from
  public TestRunner addInputStream(CollectionStream stream) {...}
  // Configures an output stream for samza system, that job can producer to
  public TestRunner addOutputStream(CollectionStream stream) {...}
   
  // Consume messages Runfrom thea appStream
  public void run(static <T> Map<Integer, List<T>> consumeStream(CollectionStream stream, Integer timeout) {...}
}

 

  
  // Run the TestRunner
  public void run() {...}
}

 

/**
* CollectionStream acts a descriptor that can be used to build an in memory input stream (single/multiple partition) of java collections(list, map).
*/
 
public class CollectionStream<T> { 
  // Create an empty stream with single partition that a Samza job can produce to
  public static <T> CollectionStream<T> empty(String systemName, String streamName) {...}
  
  // Create an aempty stream ofwith messagesmultiple frompartitions inputthat lista withSamza single partitionjob can produce to
  public static <T> CollectionStream<T> ofempty(String systemName, String streamName, Iterable<T>Integer collectionpartitionCount) {...}
  
  // Create a stream of messages from input list with multiplesingle partition, key of partitions map is partitionId
  public static <T> CollectionStream<T> of(String streamNamesystemName, String Map<IntegerstreamName,? extendsIterable<T> Iterable<T>> partitionscollection){...}
  
  // Create 
a stream publicof CollectionStream<T>messages from(CollectionStreamSystem system) {...}  
}
 input list with multiple partition, key of partitions map is partitionId
  public static <T> CollectionStream<T> of(String systemName, String streamName, Map<Integer,? extends Iterable<T>> partitions){...}
 
}

/**
* CollectionStreamSystem provides utilities to create and initialize an in memory input stream.
*/
 
public class CollectionStreamSystem { 

 // Create a CollectionStreamSystem 
 public static CollectionStreamSystem create(String name) {...}  
 // 
 public <T> CollectionStreamSystem addInput(String streamName, Map<Integer, Iterable<T>> partitions) {...}
 
 public <T> CollectionStreamSystem addOutput(String streamName, Map<Integer, Iterable<T>> partitions) {...}
 
}

/**
* Simple Test case using a collection as an input for a low level application. It demonstrates set up and comparison of a test with 
* minimal(eg. here none) configs and it reads an input stream of integers and multiplies each integer with 10
*/
 
// Create a StreamTask
MyStreamTestTask myTask = new StreamTask() {
  @Override
  public void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator) throws Exception {
    Integer obj = (Integer) envelope.getMessage();
    collector.send(new OutgoingMessageEnvelope(new SystemStream("test","output"), obj*10));
  }
};
 
CollectionStreamSystem system = CollectionStreamSystem.create("test"));
  }
};
 

CollectionStream<Integer> input = CollectionStream
	.of("test", "input", {1,2,3,4})
	.from(system);


CollectionStream output = CollectionStream
	.empty("test", "output")
	.from(system);

TestRunner
    .of(MyStreamTestTask.class)
    .addInputStream(input)
    .addOutputStream(output)
    .run();
 
 
// Assertions on the outputs
Assert.assertThat(StreamUtilsTestRunner.getStreamStateconsumeStream(output), IsIterableContainingInOrder.contains({10,20,30,40})));

/**
* Simple Test case using a collection as an input for a High level application
*/
 
public class MyStreamApplication implements StreamApplication {
  @Override
  public void init(StreamGraph graph, Config config) {
    MessageStream<Integer> pageViews = graph.getInputStream(“test-samza.page-views”);
    pageViews.map(s -> "processed " + s)
             .sendTo(graph.getOutputStream(“test-samza.output”));
 }
}
 
CollectionStreamSystem system = CollectionStreamSystem.create("test");

CollectionStream<Integer> input = CollectionStream
	.of("test", "input", {1,2,3,4})
	.from(system);

CollectionStream output = CollectionStream
	.empty("test", "output")
	.from(system);
 
// Initialize and run the test framework
TestRunner
	.of(new MyStreamApplication());
    .addInputStream(input)
    .addOutputStream(output)
    .addOverrideConfig("job.default.system", "test") 
    .run();
 
// Assertions on the outputs
StreamAssert.that(output).contains({"processed 1", "processed 2", "processed 4"});