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

@Test
public void testNodeProcessing() throws Exception
{
  InvertIndexArray<String,String> oper = new InvertIndexArray<String,String>();
  CollectorTestSink indexSink = new CollectorTestSink();
  Sink inSink = oper.data.getSink();
  oper.index.setSink(indexSink);
  oper.beginWindow(0);
  HashMap<String, ArrayList> input = new HashMap<String, ArrayList>();
  ArrayList<String> alist = new ArrayList<String>();
  alist.add("str");
  alist.add("str1");
  input.put("a", alist);
  input.put("b", alist);
  inSink.put(input);
  alist = new ArrayList<String>();
  input = new HashMap<String, ArrayList>();
  alist.add("blah");
  alist.add("str1");
  input.put("c", alist);
  inSink.put(input);
  oper.endWindow();
  Assert.assertEquals("number emitted tuples", 3, indexSink.collectedTuples.size());
  for (Object o: indexSink.collectedTuples) {
    log.debug(o.toString());
    HashMap<String, ArrayList<String>> output = (HashMap<String, ArrayList<String>>)o;
    for (Map.Entry<String, ArrayList<String>> e: output.entrySet()) {
      String key = e.getKey();
      alist = e.getValue();
      if (key.equals("str1")) {
        Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a"));
        Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b"));
        Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c"));
      } else if (key.equals("str")) {
        Assert.assertEquals("Index for \"str1\" contains \"a\"", true, alist.contains("a"));
        Assert.assertEquals("Index for \"str1\" contains \"b\"", true, alist.contains("b"));
        Assert.assertEquals("Index for \"str1\" contains \"c\"", false, alist.contains("c"));
      } else if (key.equals("blah")) {
        Assert.assertEquals("Index for \"str1\" contains \"a\"", false, alist.contains("a"));
        Assert.assertEquals("Index for \"str1\" contains \"b\"", false, alist.contains("b"));
        Assert.assertEquals("Index for \"str1\" contains \"c\"", true, alist.contains("c"));
      }
    }
  }
}