@Test
public void testProcessEvent() throws Exception {

  List mutations = Collections.singletonList(mock(Mutation.class));
  when(mapper.map(event)).thenReturn(mutations);
  when(filter.apply(event)).thenReturn(false);
  source.processEvent(event);
  verifyZeroInteractions(metrics);
  verify(listener, times(0)).onMutation(mutations);
  when(filter.apply(event)).thenReturn(true);
  source.processEvent(event);
  verify(listener, times(1)).onMutation(mutations);
  when(mapper.map(event)).thenReturn(Collections.emptyList());
  source.processEvent(event);
  verify(listener, times(2)).onEvent(event);
  verify(metrics, times(2)).eventReceived(event);
  verify(listener, times(1)).onMutation(mutations);
}