@Test
public void emitDirect() throws Exception {

    setupExpectationsForTopologyContextEmit();
    setupExpectationsForEventCorrelationInjector();
    sut = new EventCorrelatingSpoutOutputCollector(mockedTopologyContext, mockedOutputCollector);
    Deencapsulation.setField(sut, "eventCorrelationInjector", mockEventCorrelationInjector);
    String testStreamId = "testStreamId";
    final Values tuple = new Values(INPUT_STREAMLINE_EVENT);
    String messageId = "testMessageId";
    // int taskId, String streamId, List<Object> tuple
    new Expectations() {{
        mockedOutputCollector.emitDirect(TASK_1, testStreamId, withAny(tuple));
    }};
    sut.emitDirect(TASK_1, testStreamId, tuple);
    new Verifications() {{
        List<Object> capturedValues;
        mockedOutputCollector.emitDirect(TASK_1, testStreamId, capturedValues = withCapture());
        StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT);
        List<Tuple> capturedParents;
        mockEventCorrelationInjector.injectCorrelationInformation(tuple,
                capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM);
        assertEquals(0, capturedParents.size());
    }};
    // int taskId, List<Object> tuple
    new Expectations() {{
        mockedOutputCollector.emitDirect(TASK_1, withAny(tuple));
    }};
    sut.emitDirect(TASK_1, tuple);
    new Verifications() {{
        List<Object> capturedValues;
        mockedOutputCollector.emitDirect(TASK_1, capturedValues = withCapture());
        StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT);
        List<Tuple> capturedParents;
        mockEventCorrelationInjector.injectCorrelationInformation(tuple,
                capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM);
        assertEquals(0, capturedParents.size());
    }};
    // int taskId, String streamId, List<Object> tuple, Object messageId
    new Expectations() {{
        mockedOutputCollector.emitDirect(TASK_1, testStreamId, withAny(tuple), messageId);
    }};
    sut.emitDirect(TASK_1, testStreamId, tuple, messageId);
    new Verifications() {{
        List<Object> capturedValues;
        mockedOutputCollector.emitDirect(TASK_1, testStreamId, capturedValues = withCapture(), messageId);
        StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT);
        List<Tuple> capturedParents;
        mockEventCorrelationInjector.injectCorrelationInformation(tuple,
                capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM);
        assertEquals(0, capturedParents.size());
    }};
    // int taskId, List<Object> tuple, Object messageId
    new Expectations() {{
        mockedOutputCollector.emitDirect(TASK_1, withAny(tuple), messageId);
    }};
    sut.emitDirect(TASK_1, tuple, messageId);
    new Verifications() {{
        List<Object> capturedValues;
        mockedOutputCollector.emitDirect(TASK_1, capturedValues = withCapture(), messageId);
        StreamlineEventTestUtil.assertEventIsProperlyCopied((StreamlineEvent) capturedValues.get(0), INPUT_STREAMLINE_EVENT);
        List<Tuple> capturedParents;
        mockEventCorrelationInjector.injectCorrelationInformation(tuple,
                capturedParents = withCapture(), TEST_COMPONENT_NAME_FOR_STORM);
        assertEquals(0, capturedParents.size());
    }};
}