@Test
public void uniqueCountTest()
{

  UniqueValueCount<String> uniqueCountOper = new UniqueValueCount<String>();
  CollectorTestSink<KeyValPair<String, Integer>> outputSink =
      new CollectorTestSink<KeyValPair<String, Integer>>();
  CollectorTestSink<KeyValPair<String, Set<Object>>> outputSetSink =
      new CollectorTestSink<KeyValPair<String, Set<Object>>>();
  TestUtils.setSink(uniqueCountOper.output, outputSink);
  TestUtils.setSink(uniqueCountOper.outputValues, outputSetSink);
  uniqueCountOper.beginWindow(0);
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 1));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2));
  uniqueCountOper.endWindow();
  Assert.assertEquals("number emitted tuples", 1, outputSink.collectedTuples.size());
  KeyValPair<String, Integer> emittedPair = outputSink.collectedTuples.get(0);
  Assert.assertEquals("emitted key was ", "test1", emittedPair.getKey());
  Assert.assertEquals("emitted value was ", 2, emittedPair.getValue().intValue());
  Assert.assertEquals("number emitted tuples", 1, outputSetSink.collectedTuples.size());
  KeyValPair<String, Set<Object>> emittedSetPair = outputSetSink.collectedTuples.get(0);
  Assert.assertTrue(emittedSetPair.getValue().contains(1));
  Assert.assertTrue(emittedSetPair.getValue().contains(2));
  outputSink.clear();
  uniqueCountOper.beginWindow(1);
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 1));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test1", 2));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 1));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 2));
  uniqueCountOper.input.process(new KeyValPair<String, Object>("test2", 2));
  uniqueCountOper.endWindow();
  ImmutableMap<String, Integer> answers = ImmutableMap.of("test1", 2, "test2", 2);
  Assert.assertEquals("number emitted tuples", 2, outputSink.collectedTuples.size());
  for (KeyValPair<String, Integer> emittedPair2 : outputSink.collectedTuples) {
    Assert.assertEquals("emmit value of " + emittedPair2.getKey() + " was ", answers.get(emittedPair2.getKey()),
        emittedPair2.getValue());
  }
  LOG.debug("Done unique count testing testing\n");
}