@SuppressWarnings({ "rawtypes", "unchecked" })

@Test
public void testNodeProcessing() throws Exception
{
  UniqueCounter<String> oper = new UniqueCounter<String>();
  CountAndLastTupleTestSink sink = new CountAndLastTupleTestSink<HashMap<String, Integer>>();
  oper.count.setSink(sink);
  String atuple = "a";
  String btuple = "b";
  String ctuple = "c";
  String dtuple = "d";
  int numTuples = 10000;
  oper.beginWindow(0);
  for (int i = 0; i < numTuples; i++) {
    oper.data.process(atuple);
    if (i % 2 == 0) {
      oper.data.process(btuple);
    }
    if (i % 5 == 0) {
      oper.data.process(ctuple);
    }
  }
  oper.endWindow();
  oper.beginWindow(1);
  for (int i = 0; i < numTuples; i++) {
    oper.data.process(atuple);
  }
  oper.endWindow();
  HashMap<String, Integer> tuple = (HashMap<String, Integer>)sink.tuple;
  int acount = tuple.get("a");
  Assert.assertEquals("number emitted tuples", numTuples, acount);
  oper.beginWindow(2);
  for (int i = 0; i < numTuples; i++) {
    if (i % 2 == 0) {
      oper.data.process(btuple);
    }
    oper.data.process(btuple);
    if (i % 10 == 0) {
      oper.data.process(dtuple);
    }
  }
  oper.endWindow();
  tuple = (HashMap<String, Integer>)sink.tuple;
  int bcount = tuple.get("b");
  int dcount = tuple.get("d");
  Assert.assertEquals("number emitted tuples", 3, sink.count);
  Assert.assertEquals("number emitted tuples", numTuples + (numTuples / 2), bcount);
  Assert.assertEquals("number emitted tuples", numTuples / 10, dcount);
}