@Test
public void testExecute() {

    Message message = new Message(mentionTime, "randomFrom", "randomUserId",
                                  String.format("test message with :%s:", emoji),
                                  "randomRoomId", MessageType.MESSAGE);
    FatMessage fatMessage = new FatMessage(message, user, room);
    List<Object> values = Lists.newArrayList(fatMessage);
    TopologyContext context = mock(TopologyContext.class);
    Fields fields = mock(Fields.class);
    when(fields.size()).thenReturn(1);
    when(context.getComponentOutputFields(anyString(), anyString())).thenReturn(fields);
    Tuple input = new TupleImpl(context, values, 0, "stream-id");
    BasicOutputCollector collector = mock(BasicOutputCollector.class);
    ChatAlyticsConfig config = new ChatAlyticsConfig();
    config.computeConfig.apiRetries = 0;
    config.persistenceUnitName = "chatalytics-db-test";
    Map<Object, Object> stormConf = Maps.newHashMapWithExpectedSize(1);
    stormConf.put(ConfigurationConstants.CHATALYTICS_CONFIG.txt, YamlUtils.writeYaml(config));
    underTest.prepare(stormConf, context);
    underTest.execute(input, collector);
    verify(collector).emit(any(Values.class));
}