MUnit testing a flow with collection-aggregator

Question

I am trying to write MUnit tests for my Mule flows. I want to write a test for the below flow

<flow name="DownloadFTPFileIntoLocalFlow" processingStrategy="synchronous" tracking:enable-default-events="true">
    <quartz:inbound-endpoint jobName="Source-File-Scheduler" cronExpression="${source.pollingfrequency}" startDelay="10000" responseTimeout="10000" doc:name="Quartz">
        <quartz:endpoint-polling-job>
            <quartz:job-endpoint ref="InputSFTPEndpoint"/>
        </quartz:endpoint-polling-job>
    </quartz:inbound-endpoint>
    <logger message="DownloadFTPFileIntoLocalFlow #[payload.getClass().getName()]" level="INFO" doc:name="Logger"/>
    <set-property propertyName="MULE_CORRELATION_GROUP_SIZE" value="#[java.lang.Integer.MAX_VALUE]" doc:name="GroupsizeForExceptionAggregator"/>
    <set-property propertyName="MULE_CORRELATION_ID" value="#[java.util.UUID.randomUUID().toString()]" doc:name="corelationIdForExceptionAggregator"/>
    <set-variable variableName="originalPayload" value="#[payload]" doc:name="originalPayload"/>
    <byte-array-to-object-transformer doc:name="Byte Array to Object"/>
    <flow-ref name="ProcessCSVFlow" doc:name="ProcessCSVFlow" />
    <exception-strategy ref="Default_Exception_Strategy" doc:name="DownloadFTPFileIntoLocalFlow Strategy"/>
</flow>


 <sub-flow name="ProcessCSVFlow" tracking:enable-default-events="true">
    <transformer ref="enrichWithHeaderAndEndOfFileTransformer" doc:name="headerAndEOFEnricher" />
    <set-variable variableName="outputfilename" value="#['Mercury'+server.dateTime.year+server.dateTime.month+server.dateTime.dayOfMonth+server.dateTime.hours+server.dateTime.minutes+server.dateTime.seconds+'.csv']" doc:name="outputfilename"/>
    <!-- <set-variable variableName="outputfilename" value="#['Mercury'+server.dateTime.year+':'+server.dateTime.month+':'+server.dateTime.dayOfMonth+'::'+server.dateTime.hours+':'+server.dateTime.minutes+':'+server.dateTime.seconds+'.csv']" doc:name="outputfilename"/> -->
    <sftp:outbound-endpoint exchange-pattern="one-way" connector-ref="DestinationSFTP" host="${destination.host}" port="22" responseTimeout="10000" doc:name="DestinationSFTP"
    outputPattern="#[outputfilename]" path="${destination.path}" user="${destination.username}" password="${destination.password}"/>
    <gzip-compress-transformer/>
    <sftp:outbound-endpoint exchange-pattern="one-way" connector-ref="InputSFTP" host="${source.host}" port="22" responseTimeout="10000" doc:name="SourceArchiveSFTP" 
    outputPattern="#[outputfilename].gzip" path="Archive" user="${source.username}" password="${source.password}"/>
    <component doc:name="Delete Read File">
        <singleton-object class="component.DeleteProcessedFileComponent">
            <property key="host" value="${source.host}"/>
            <property key="username" value="${source.username}"/>
            <property key="password" value="${source.password}"/>
            <property key="workingDirectory" value="${source.path}"/>
        </singleton-object>
    </component>
    <parse-template location="successmessagetemplate.txt" doc:name="Success Template"/>
    <smtp:outbound-endpoint host="${smtp.host}" port="${smtp.port}" user="${smtp.from.address}" password="${smtp.from.password}" 
                            to="${smtp.to.address}" from="${smtp.from.address}" subject="${mail.success.subject}" responseTimeout="10000" 
                            doc:name="SuccessEmail" connector-ref="Gmail"/>
    <logger message="Process completed successfully" level="INFO" doc:name="Logger"/>
</sub-flow>

Exception handling block

<catch-exception-strategy name="Default_Exception_Strategy">
    <flow-ref name="ExceptionHandlingSubflow" doc:name="ExceptionHandlingSubflow"/>
</catch-exception-strategy>

<sub-flow name="ExceptionHandlingSubflow" tracking:enable-default-events="true">
    <collection-aggregator timeout="60000" failOnTimeout="false" doc:name="Exception Aggregator"/>
    <logger message="Exception has occured Payload is #[payload] and Message is #[message]" level="ERROR" doc:name="Logger"/>
    <parse-template location="errormessagetemplate.txt" doc:name="Error Template"/>
    <smtp:outbound-endpoint host="${smtp.host}" port="${smtp.port}" user="${smtp.from.address}" password="${smtp.from.password}"
                            to="${smtp.to.address}" from="${smtp.from.address}" subject="${mail.failure.subject}" responseTimeout="10000" 
                            doc:name="ErrorEmail" connector-ref="Gmail"/>
</sub-flow>

The interesting bit is the exception sub-flow, especially the collection-aggregator

My unit test is

@Test
public void whenMultipleExceptionsOccurInFlow_itShouldSendOnlyOneFailureEmail() throws Exception {
    whenMessageProcessor("collection-aggregator")
            .withAttributes(attribute("name").ofNamespace("doc").withValue("Exception Aggregator")).thenReturnSameEvent();

    destinationSFTP.thenThrow(new RuntimeException("Dummy Exception destinationSFTP"));

    MuleEvent testEvent = PropertyEnricher.enrich(testEvent(IOUtils.toInputStream("hello,dummy,payload"))).get();

    runFlow("DownloadFTPFileIntoLocalFlow", testEvent);

    verifyCallOfMessageProcessor("outbound-endpoint").ofNamespace("smtp")
            .withAttributes(attribute("name").ofNamespace("doc").withValue("ErrorEmail"))
            .times(1);
}

Now if I do not mock the collection aggregator out my test does not pass, I can understand that this is tricky as the aggregator has a "pause" within it and hence is not an ideal candidate for a unit test, however from a technical standpoint I want to understand what is causing the unit test to fail (when collection-aggregator is not mocked).

My test fails when the collection-aggregator is not mocked.

junit.framework.AssertionFailedError: On smtp:outbound-endpoint.Expected 1 but got 0 calls
    at junit.framework.Assert.fail(Assert.java:50)
    at org.mule.munit.common.mocking.MunitVerifier.times(MunitVerifier.java:86)
    at nz.co.mightyriver.ProcessCsvTest.whenMultipleExceptionsOccurInFlow_itShouldSendOnlyOneFailureEmail(ProcessCsvTest.java:100)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)

Answer 1

So I've been trying to reproduce this issue:

Production Code

<http:listener-config name="HTTP_Listener_Configuration" host="0.0.0.0" port="9090" doc:name="HTTP Listener Configuration"/>
    <flow name="stack-munit-and-aggregationFlow">
        <http:listener config-ref="HTTP_Listener_Configuration" path="/" doc:name="HTTP"/>
        <set-payload value="#['lalero_' + new java.util.Date().toString()]" doc:name="Set Payload"/>
        <flow-ref name="stack-munit-and-aggregationSub_Flow" doc:name="stack-munit-and-aggregationSub_Flow"/>
        <set-payload doc:name="Set Payload" value="#[payload.toString()]"/>
    </flow>
    <sub-flow name="stack-munit-and-aggregationSub_Flow">
        <collection-aggregator failOnTimeout="true" doc:name="Collection Aggregator" timeout="10"/>
    </sub-flow>

Test Code

package org.mule.munit;

import org.junit.Assert; import org.junit.Test; import org.mule.api.MuleEvent; import org.mule.api.MuleException; import org.mule.munit.runner.functional.FunctionalMunitSuite;

public class TheTest extends FunctionalMunitSuite {

    @Test
    public void aTest() throws MuleException, Exception {    MuleEvent event = runFlow("stack-munit-and-aggregationFlow", testEvent(""));

    String payload = (String) event.getMessage().getPayload();

    Assert.assertTrue(payload.contains("lalero"));
    } }

If you check this code you'll notice that I'm not mocking out the collection aggregator. After a few tests I wasn't able to reproduce your error.

I think the issue could somewhere else. Could you please share you're code so I can investigate further?

World of warning though, due to an issue discovered in: How to mock a Java component within Mule Flow using MUnit

You may find a problem if you try to directly test your sub-flow ExceptionHandlingSubFlow. But as you were not doing that in your example code I don't think those two are related.

Cheers!