Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/kstream/internals/KStreamFlatTransformValuesTest.java (Java)

Code before:
import java.util.Collections;

import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.kstream.internals.KStreamFlatTransformValues.KStreamFlatTransformValuesProcessor;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class KStreamFlatTransformValuesTest {

    private Integer inputKey;
    private Integer inputValue;

    @Mock
    private ValueTransformerWithKey<Integer, Integer, Iterable<String>> valueTransformer;
    @Mock
    private InternalProcessorContext<Integer, String> context;

    private KStreamFlatTransformValuesProcessor<Integer, Integer, String> processor;

    @Before
    public void setUp() {
        inputKey = 1;
        inputValue = 10;
        processor = new KStreamFlatTransformValuesProcessor<>(valueTransformer);
    }

    @Test
    public void shouldInitializeFlatTransformValuesProcessor() {
        processor.init(context);

        verify(valueTransformer).init(ArgumentMatchers.isA(ForwardingDisabledProcessorContext.class));
    }

    @Test
    public void shouldTransformInputRecordToMultipleOutputValues() {
        final Iterable<String> outputValues = Arrays.asList(
                "Hello",
                "Blue",
                "Planet");

        processor.init(context);

Review comment (suggestion):
n EasyMock a strict mock verifies also the order of the calls. In Mockito that is done by using `InOrder` (https://javadoc.io/doc/org.mockito/mockito-core/latest/org/mockito/Mockito.html#in_order_verification). Currently, this PR does not check the order which softens up the test. 

Code after:
import java.util.Arrays;
import java.util.Collections;

import org.apache.kafka.streams.kstream.ValueTransformerWithKey;
import org.apache.kafka.streams.kstream.ValueTransformerWithKeySupplier;
import org.apache.kafka.streams.kstream.internals.KStreamFlatTransformValues.KStreamFlatTransformValuesProcessor;
import org.apache.kafka.streams.processor.api.Processor;
import org.apache.kafka.streams.processor.api.Record;
import org.apache.kafka.streams.processor.internals.ForwardingDisabledProcessorContext;
import org.apache.kafka.streams.processor.internals.InternalProcessorContext;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.ArgumentMatchers;
import org.mockito.InOrder;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;

@RunWith(MockitoJUnitRunner.StrictStubs.class)
public class KStreamFlatTransformValuesTest {

    private Integer inputKey;
    private Integer inputValue;

    @Mock
    private ValueTransformerWithKey<Integer, Integer, Iterable<String>> valueTransformer;
    @Mock
    private InternalProcessorContext<Integer, String> context;
    private InOrder inOrder;

    private KStreamFlatTransformValuesProcessor<Integer, Integer, String> processor;

    @Before
    public void setUp() {
        inputKey = 1;
        inputValue = 10;
        inOrder = inOrder(context);
        processor = new KStreamFlatTransformValuesProcessor<>(valueTransformer);
    }

    @Test
    public void shouldInitializeFlatTransformValuesProcessor() {
        processor.init(context);

        verify(valueTransformer).init(ArgumentMatchers.isA(ForwardingDisabledProcessorContext.class));
    }

    @Test
    public void shouldTransformInputRecordToMultipleOutputValues() {
        final Iterable<String> outputValues = Arrays.asList(
                "Hello",