@Test
public void testAddOperator()
{

  LogicalPlan dag = new LogicalPlan();
  TestOperator<String, Integer> firstOperator = new TestOperator<>();
  TestOperator<Integer, Date> secondOperator = new TestOperator<>();
  new ApexStreamImpl<String>().addOperator(firstOperator, null, firstOperator.output, name("first"))
      .endWith(secondOperator, secondOperator.input, name("second"))
      .with(DAG.Locality.THREAD_LOCAL)
      .with(Context.OperatorContext.AUTO_RECORD, true)
      .with("prop", "TestProp").populateDag(dag);
  Assert.assertTrue(dag.getAllOperators().size() == 2);
  Set<String> opNames = new HashSet<>();
  opNames.add("first");
  opNames.add("second");
  for (LogicalPlan.OperatorMeta operatorMeta : dag.getAllOperators()) {
    Assert.assertTrue(operatorMeta.getOperator() instanceof TestOperator);
    Assert.assertTrue(opNames.contains(operatorMeta.getName()));
    if (operatorMeta.getName().equals("second")) {
      // Assert the Context.OperatorContext.AUTO_RECORD attribute has been set to true for second operator
      Assert.assertTrue(operatorMeta.getAttributes().get(Context.OperatorContext.AUTO_RECORD));
      // Assert the prop has been set to TestProp for second operator
      Assert.assertEquals("TestProp", ((TestOperator)operatorMeta.getOperator()).prop);
    } else {
      // Assert the Context.OperatorContext.AUTO_RECORD attribute keeps as default false for first operator
      Assert.assertNull(operatorMeta.getAttributes().get(Context.OperatorContext.AUTO_RECORD));
      // Assert the prop has not been set for first operator
      Assert.assertNull(((TestOperator)operatorMeta.getOperator()).prop);
    }
  }
  Collection<LogicalPlan.StreamMeta> streams = dag.getAllStreams();
  // Assert there is only one stream
  Assert.assertTrue(streams.size() == 1);
  for (LogicalPlan.StreamMeta stream : streams) {
    // Assert the stream is from first operator to second operator
    Assert.assertEquals("first", stream.getSource().getOperatorMeta().getName());
    Collection<InputPortMeta> portMetaCollection = stream.getSinks();
    Assert.assertTrue(1 == portMetaCollection.size());
    for (InputPortMeta inputPortMeta : portMetaCollection) {
      Assert.assertEquals("second", inputPortMeta.getOperatorMeta().getName());
    }
    // Assert the stream is thread local
    Assert.assertTrue(stream.getLocality() == DAG.Locality.THREAD_LOCAL);
  }
}