MUnit 使用 collection-aggregator 测试流

Question

我正在尝试为我的 Mule 流编写 MUnit 测试。我想为下面的流程写一个测试

<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>

异常处理块

<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>

有趣的是异常子流，尤其是集合聚合器

我的单元测试是

@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);
}

现在，如果我不模拟集合聚合器，我的测试没有通过，我可以理解这很棘手，因为聚合器中有一个“暂停”，因此不是单元测试的理想候选者，但是从一个技术角度我想了解是什么导致单元测试失败（当集合聚合器未被模拟时）。

当集合聚合器未被模拟时，我的测试失败。

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

所以我一直在尝试重现这个问题：

生产代码

<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>

测试代码

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"));
    } }

如果您检查此代码，您会注意到我并没有在模拟集合聚合器。经过几次测试，我无法重现您的错误。

我认为问题可能在其他地方。 您能否分享您的代码，以便我进一步调查？

警告世界，但由于在以下位置发现了一个问题： How to mock a Java component within Mule Flow using MUnit

如果您尝试直接测试您的子流 ExceptionHandlingSubFlow，您可能会发现问题。但是由于您在示例代码中没有这样做，我认为这两者不相关。

干杯！