/**
* 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 and AsyncStreamTask
*/ 

public class TestRunner {
  
  // Static Factory to config & create runner for low level api 
  public static TestRunner of(StreamTask task) {...}
  public static TestRunner of(AsyncStreamTask task) {...}
  public static TestRunner of(StreamApplication app) {...}
  
  // Add/Ovveride any custom configs
  public TestRunner addOverrideConfigs(Map<String,String> configs) {...}
  public TestRunner addOverrideConfigs(String configUri) {...}
  public TestRunner addConfigaddOverrideConfig(String key, String val){...} 
  // Set container mode either single container or multi container
  public TestRunner setContainerMode(Mode mode) {...}
  
  // Multithreading configs for users
  public TestRunner setTaskMaxConcurrency(Integer value) {...}
  public TestRunner setTaskCallBackTimeoutMS(Integer value) {...}
  public TestRunner setJobContainerThreadPoolSize(Integer value) {...}
 
  // Configure state for application
  public TestRunner addState(String storeName) {...}  
  
  // Configure an input stream for samza system, that task can consume from
  public TestRunner addInputStream(CollectionStream stream) {...}
  // Configures an output stream for samza system, that task can producer to
  public TestRunner addOutputStream(CollectionStream stream) {...}
  
  // Run the app
  public void run() {...}
}

 

/**
* 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”));
 }
}


 
// Initialize and run the test framework
TestRunner
	.of(new MyStreamApplication());
    .addInputStream(CollectionStream.of("test-samza.input", {1,2,3}))
    .addOutputStream(CollectionStream.empty("test-samza.Outputoutput"))
    .run();
 
// Assertions on the outputs
StreamAssert.that("test-samza.input").contains(Arrays.asList({"processed 1", "processed 2", "processed 4")});