HistoryBotRemoverTest

@Test
public void testHistoryBotRemover() {
  Fields pageViewFields = new Fields("id", "pageViewId", "path", "campaignTag", "sessionId", "userId", "remoteHost",
      "host");
  Fields sessionFields = new Fields("sessionId", "campaignTag");
  TupleInputTap campaignPageViewsTap = new TupleInputTap("campaignPageViewsTap",
      pageViewFields,
      new Tuple(0L, "0", "/foo", "bar", 0L, 0L, "1.1.1.1", "foo.bar.com"),
      new Tuple(3L, "3", "/foo", "bar", 1L, 0L, "1.1.1.1", "foo.bar.com"),
  AvroInMemoryInputTap<UserEvent> userEventsTap = new AvroInMemoryInputTap<UserEvent>(UserEvent.getClassSchema(),
      userEvent().setId(0).setPageViewId("0").setCategory("pageload").build());
  TupleOutputTap nonBotPageViewsTap = new TupleOutputTap("nonBotPageViewsTap", pageViewFields);
  TupleOutputTap nonBotSessionsTap = new TupleOutputTap("nonBotPageViewsTap", sessionFields);

  nonBotPageViews = new HistoryBotRemover(campaignPageViews, userEvents);

  Flow<Properties> flow = new LocalFlowConnector().connect(new FlowDef()
      .addSource(campaignPageViews, campaignPageViewsTap)
      .addSource(userEvents, userEventsTap)
      .addTailSink(nonBotPageViews, nonBotPageViewsTap));
  flow.complete();

  Set<Tuple> expectedPageViews = Sets.newHashSet(
      new Tuple(0L, "0", "/foo", "bar", 0L, 0L, "1.1.1.1", "foo.bar.com"));
  assertEquals(expectedPageViews, nonBotPageViewsTap.getResults());
}