@Test
public void testStatsListenerContextWrappers()
{

  LogicalPlan dag = new LogicalPlan();
  dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent());
  StatsListenerOperator o1 = dag.addOperator("o1", new StatsListenerOperator());
  GenericTestOperator o2 = dag.addOperator("o2", new GenericTestOperator());
  dag.setAttribute(o2, OperatorContext.STATS_LISTENERS,
      Lists.<StatsListener>newArrayList(mock(StatsListener.class)));
  GenericTestOperator o3 = dag.addOperator("o3", new GenericTestOperator());
  dag.setAttribute(o3, OperatorContext.STATS_LISTENERS,
      Lists.<StatsListener>newArrayList(mock(StatsListenerWithContext.class)));
  StatsListenerOperatorOld o4 = dag.addOperator("o4", new StatsListenerOperatorOld());
  PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext());
  PTOperator p1 = plan.getOperators(dag.getMeta(o1)).get(0);
  StatsListener l = p1.statsListeners.get(0);
  Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext);
  PTOperator p2 = plan.getOperators(dag.getMeta(o2)).get(0);
  l = p1.statsListeners.get(0);
  Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext);
  PTOperator p3 = plan.getOperators(dag.getMeta(o3)).get(0);
  l = p1.statsListeners.get(0);
  Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext);
  PTOperator p4 = plan.getOperators(dag.getMeta(o4)).get(0);
  l = p1.statsListeners.get(0);
  Assert.assertTrue("Operator stats listener is wrapped ", l instanceof StatsListenerWithContext);
}