Java example source code file (OpTestCase.java)

This example Java source code file (OpTestCase.java) is included in the alvinalexander.com "Java Source Code Warehouse" project. The intent of this project is to help you "Learn Java by Example" ^TM.

Learn more about this Java project at its project page.

Java - Java tags/keywords

abstractpipeline, basestream, collection, exercisedatastreambuilder, exercisedataterminalbuilder, function, override, s_in, s_out, streamshape, suppresswarnings, util

The OpTestCase.java Java example source code

/*
 * Copyright (c) 2012, 2013, Oracle and/or its affiliates. All rights reserved.
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * This code is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License version 2 only, as
 * published by the Free Software Foundation.
 *
 * This code is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License
 * version 2 for more details (a copy is included in the LICENSE file that
 * accompanied this code).
 *
 * You should have received a copy of the GNU General Public License version
 * 2 along with this work; if not, write to the Free Software Foundation,
 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
 *
 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
 * or visit www.oracle.com if you need additional information or have any
 * questions.
 */
package java.util.stream;

import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumMap;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.Spliterator;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import java.util.function.Function;

import org.testng.annotations.Test;

/**
 * Base class for streams test cases.  Provides 'exercise' methods for taking
 * lambdas that construct and modify streams, and evaluates them in different
 * ways and asserts that they produce equivalent results.
 */
@Test
public abstract class OpTestCase extends LoggingTestCase {

    private final Map<StreamShape, Set testScenarios;

    protected OpTestCase() {
        testScenarios = new EnumMap<>(StreamShape.class);
        testScenarios.put(StreamShape.REFERENCE, Collections.unmodifiableSet(EnumSet.allOf(StreamTestScenario.class)));
        testScenarios.put(StreamShape.INT_VALUE, Collections.unmodifiableSet(EnumSet.allOf(IntStreamTestScenario.class)));
        testScenarios.put(StreamShape.LONG_VALUE, Collections.unmodifiableSet(EnumSet.allOf(LongStreamTestScenario.class)));
        testScenarios.put(StreamShape.DOUBLE_VALUE, Collections.unmodifiableSet(EnumSet.allOf(DoubleStreamTestScenario.class)));
    }

    @SuppressWarnings("rawtypes")
    public static int getStreamFlags(BaseStream s) {
        return ((AbstractPipeline) s).getStreamFlags();
    }

    /**
     * An asserter for results produced when exercising of stream or terminal
     * tests.
     *
     * @param <R> the type of result to assert on
     */
    public interface ResultAsserter<R> {
        /**
         * Assert a result produced when exercising of stream or terminal
         * test.
         *
         * @param actual the actual result
         * @param expected the expected result
         * @param isOrdered true if the pipeline is ordered
         * @param isParallel true if the pipeline is parallel
         */
        void assertResult(R actual, R expected, boolean isOrdered, boolean isParallel);
    }

    // Exercise stream operations

    public interface BaseStreamTestScenario {
        StreamShape getShape();

        boolean isParallel();

        abstract <T, U, S_IN extends BaseStream>
        void run(TestData<T, S_IN> data, Consumer b, Function m);
    }

    public <T, U, S_IN extends BaseStream>
    Collection<U> exerciseOps(TestData data, Function m) {
        return withData(data).stream(m).exercise();
    }

    // Run multiple versions of exercise(), returning the result of the first, and asserting that others return the same result
    // If the first version is s -> s.foo(), can be used with s -> s.mapToInt(i -> i).foo().mapToObj(i -> i) to test all shape variants
    @SafeVarargs
    public final<T, U, S_IN extends BaseStream>
    Collection<U> exerciseOpsMulti(TestData data,
                                   Function<S_IN, S_OUT>... ms) {
        Collection<U> result = null;
        for (Function<S_IN, S_OUT> m : ms) {
            if (result == null)
                result = withData(data).stream(m).exercise();
            else {
                Collection<U> r2 = withData(data).stream(m).exercise();
                assertEquals(result, r2);
            }
        }
        return result;
    }

    // Run multiple versions of exercise() for an Integer stream, returning the result of the first, and asserting that others return the same result
    // Automates the conversion between Stream<Integer> and {Int,Long,Double}Stream and back, so client sites look like you are passing the same
    // lambda four times, but in fact they are four different lambdas since they are transforming four different kinds of streams
    public final
    Collection<Integer> exerciseOpsInt(TestData.OfRef data,
                                       Function<Stream> mRef,
                                       Function<IntStream, IntStream> mInt,
                                       Function<LongStream, LongStream> mLong,
                                       Function<DoubleStream, DoubleStream> mDouble) {
        @SuppressWarnings({ "rawtypes", "unchecked" })
        Function<Stream>[] ms = new Function[4];
        ms[0] = mRef;
        ms[1] = s -> mInt.apply(s.mapToInt(e -> e)).mapToObj(e -> e);
        ms[2] = s -> mLong.apply(s.mapToLong(e -> e)).mapToObj(e -> (int) e);
        ms[3] = s -> mDouble.apply(s.mapToDouble(e -> e)).mapToObj(e -> (int) e);
        return exerciseOpsMulti(data, ms);
    }

    public <T, U, S_OUT extends BaseStream
    Collection<U> exerciseOps(Collection data, Function, S_OUT> m) {
        TestData.OfRef<T> data1 = TestData.Factory.ofCollection("Collection of type " + data.getClass().getName(), data);
        return withData(data1).stream(m).exercise();
    }

    public <T, U, S_OUT extends BaseStream>
    Collection<U> exerciseOps(Collection data, Function, S_OUT> m, I expected) {
        TestData.OfRef<T> data1 = TestData.Factory.ofCollection("Collection of type " + data.getClass().getName(), data);
        return withData(data1).stream(m).expectedResult(expected).exercise();
    }

    @SuppressWarnings("unchecked")
    public <U, S_OUT extends BaseStream
    Collection<U> exerciseOps(int[] data, Function m) {
        return withData(TestData.Factory.ofArray("int array", data)).stream(m).exercise();
    }

    public Collection<Integer> exerciseOps(int[] data, Function m, int[] expected) {
        TestData.OfInt data1 = TestData.Factory.ofArray("int array", data);
        return withData(data1).stream(m).expectedResult(expected).exercise();
    }

    public <T, S_IN extends BaseStream DataStreamBuilder withData(TestData data) {
        Objects.requireNonNull(data);
        return new DataStreamBuilder<>(data);
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    public class DataStreamBuilder<T, S_IN extends BaseStream {
        final TestData<T, S_IN> data;

        private DataStreamBuilder(TestData<T, S_IN> data) {
            this.data = Objects.requireNonNull(data);
        }

        public <U, S_OUT extends BaseStream
        ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> ops(IntermediateTestOp... ops) {
            return new ExerciseDataStreamBuilder<>(data, (S_IN s) -> (S_OUT) chain(s, ops));
        }

        public <U, S_OUT extends BaseStream ExerciseDataStreamBuilder
        stream(Function<S_IN, S_OUT> m) {
            return new ExerciseDataStreamBuilder<>(data, m);
        }

        public <U, S_OUT extends BaseStream ExerciseDataStreamBuilder
        stream(Function<S_IN, S_OUT> m, IntermediateTestOp additionalOp) {
            return new ExerciseDataStreamBuilder<>(data, s -> (S_OUT) chain(m.apply(s), additionalOp));
        }

        public <R> ExerciseDataTerminalBuilder
        terminal(Function<S_IN, R> terminalF) {
            return new ExerciseDataTerminalBuilder<>(data, s -> s, terminalF);
        }

        public <U, R, S_OUT extends BaseStream ExerciseDataTerminalBuilder
        terminal(Function<S_IN, S_OUT> streamF, Function terminalF) {
            return new ExerciseDataTerminalBuilder<>(data, streamF, terminalF);
        }
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    public class ExerciseDataStreamBuilder<T, U, S_IN extends BaseStream> {
        final TestData<T, S_IN> data;
        final Function<S_IN, S_OUT> m;
        final StreamShape shape;

        Set<BaseStreamTestScenario> testSet = new HashSet<>();

        Collection<U> refResult;

        Consumer<TestData before = LambdaTestHelpers.bEmpty;

        Consumer<TestData after = LambdaTestHelpers.bEmpty;

        ResultAsserter<Iterable resultAsserter = (act, exp, ord, par) -> {
            if (par & !ord) {
                LambdaTestHelpers.assertContentsUnordered(act, exp);
            }
            else {
                LambdaTestHelpers.assertContentsEqual(act, exp);
            }
        };

        private ExerciseDataStreamBuilder(TestData<T, S_IN> data, Function m) {
            this.data = data;

            this.m = Objects.requireNonNull(m);

            this.shape = ((AbstractPipeline<?, U, ?>) m.apply(data.stream())).getOutputShape();

            // Have to initiate from the output shape of the last stream
            // This means the stream mapper is required first rather than last
            testSet.addAll(testScenarios.get(shape));
        }

        //

        public <I extends Iterable ExerciseDataStreamBuilder expectedResult(I expectedResult) {
            List<U> l = new ArrayList<>();
            expectedResult.forEach(l::add);
            refResult = l;
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> expectedResult(int[] expectedResult) {
            List l = new ArrayList();
            for (int anExpectedResult : expectedResult) {
                l.add(anExpectedResult);
            }
            refResult = l;
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> expectedResult(long[] expectedResult) {
            List l = new ArrayList();
            for (long anExpectedResult : expectedResult) {
                l.add(anExpectedResult);
            }
            refResult = l;
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> expectedResult(double[] expectedResult) {
            List l = new ArrayList();
            for (double anExpectedResult : expectedResult) {
                l.add(anExpectedResult);
            }
            refResult = l;
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> before(Consumer> before) {
            this.before = Objects.requireNonNull(before);
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> after(Consumer> after) {
            this.after = Objects.requireNonNull(after);
            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> without(BaseStreamTestScenario... tests) {
            return without(Arrays.asList(tests));
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> without(Collection tests) {
            for (BaseStreamTestScenario ts : tests) {
                if (ts.getShape() == shape) {
                    testSet.remove(ts);
                }
            }

            if (testSet.isEmpty()) {
                throw new IllegalStateException("Test scenario set is empty");
            }

            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> with(BaseStreamTestScenario... tests) {
            return with(Arrays.asList(tests));
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> with(Collection tests) {
            testSet = new HashSet<>();

            for (BaseStreamTestScenario ts : tests) {
                if (ts.getShape() == shape) {
                    testSet.add(ts);
                }
            }

            if (testSet.isEmpty()) {
                throw new IllegalStateException("Test scenario set is empty");
            }

            return this;
        }

        public ExerciseDataStreamBuilder<T, U, S_IN, S_OUT> resultAsserter(ResultAsserter> resultAsserter) {
            this.resultAsserter = resultAsserter;
            return this;
        }

        // Build method

        public Collection<U> exercise() {
            final boolean isOrdered;
            if (refResult == null) {
                // Induce the reference result
                before.accept(data);
                S_OUT sOut = m.apply(data.stream());
                isOrdered = StreamOpFlag.ORDERED.isKnown(((AbstractPipeline) sOut).getStreamFlags());
                Node<U> refNodeResult = ((AbstractPipeline) sOut).evaluateToArrayNode(size -> (U[]) new Object[size]);
                refResult = LambdaTestHelpers.toBoxedList(refNodeResult.spliterator());
                after.accept(data);
            }
            else {
                S_OUT sOut = m.apply(data.stream());
                isOrdered = StreamOpFlag.ORDERED.isKnown(((AbstractPipeline) sOut).getStreamFlags());
            }

            List<Error> errors = new ArrayList<>();
            for (BaseStreamTestScenario test : testSet) {
                try {
                    before.accept(data);

                    List<U> result = new ArrayList<>();
                    test.run(data, LambdaTestHelpers.<U>toBoxingConsumer(result::add), m);

                    Runnable asserter = () -> resultAsserter.assertResult(result, refResult, isOrdered, test.isParallel());

                    if (refResult.size() > 1000) {
                        LambdaTestHelpers.launderAssertion(
                                asserter,
                                () -> String.format("%n%s: [actual size=%d] != [expected size=%d]", test, result.size(), refResult.size()));
                    }
                    else {
                        LambdaTestHelpers.launderAssertion(
                                asserter,
                                () -> String.format("%n%s: [actual] %s != [expected] %s", test, result, refResult));
                    }

                    after.accept(data);
                } catch (Throwable t) {
                    errors.add(new Error(String.format("%s: %s", test, t), t));
                }
            }

            if (!errors.isEmpty()) {
                StringBuilder sb = new StringBuilder();
                int i = 1;
                for (Error t : errors) {
                    sb.append(i++).append(": ");
                    if (t instanceof AssertionError) {
                        sb.append(t).append("\n");
                    }
                    else {
                        StringWriter sw = new StringWriter();
                        PrintWriter pw = new PrintWriter(sw);

                        t.getCause().printStackTrace(pw);
                        pw.flush();
                        sb.append(t).append("\n").append(sw);
                    }
                }
                sb.append("--");

                fail(String.format("%d failure(s) for test data: %s\n%s", i - 1, data.toString(), sb));
            }

            return refResult;
        }
    }

    // Exercise terminal operations

    interface BaseTerminalTestScenario<U, R, S_OUT extends BaseStream {
        boolean requiresSingleStageSource();

        boolean requiresParallelSource();

        default R run(Function<S_OUT, R> terminalF, S_OUT source, StreamShape shape) {
            return terminalF.apply(source);
        }
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    static enum TerminalTestScenario implements BaseTerminalTestScenario {
        SINGLE_SEQUENTIAL(true, false),

        SINGLE_SEQUENTIAL_SHORT_CIRCUIT(true, false) {
            @Override
            public Object run(Function terminalF, BaseStream source, StreamShape shape) {
                source = (BaseStream) chain(source, new ShortCircuitOp(shape));
                return terminalF.apply(source);
            }
        },

        SINGLE_PARALLEL(true, true),

        ALL_SEQUENTIAL(false, false),

        ALL_SEQUENTIAL_SHORT_CIRCUIT(false, false) {
            @Override
            public Object run(Function terminalF, BaseStream source, StreamShape shape) {
                source = (BaseStream) chain(source, new ShortCircuitOp(shape));
                return terminalF.apply(source);
            }
        },

        ALL_PARALLEL(false, true),

        ALL_PARALLEL_SEQUENTIAL(false, false) {
            @Override
            public Object run(Function terminalF, BaseStream source, StreamShape shape) {
                return terminalF.apply(source.sequential());
            }
        },
        ;

        private final boolean requiresSingleStageSource;
        private final boolean isParallel;

        TerminalTestScenario(boolean requiresSingleStageSource, boolean isParallel) {
            this.requiresSingleStageSource = requiresSingleStageSource;
            this.isParallel = isParallel;
        }

        @Override
        public boolean requiresSingleStageSource() {
            return requiresSingleStageSource;
        }

        @Override
        public boolean requiresParallelSource() {
            return isParallel;
        }

    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    public class ExerciseDataTerminalBuilder<T, U, R, S_IN extends BaseStream> {
        final TestData<T, S_IN> data;
        final Function<S_IN, S_OUT> streamF;
        final Function<S_OUT, R> terminalF;

        R refResult;

        ResultAsserter<R> resultAsserter = (act, exp, ord, par) -> LambdaTestHelpers.assertContentsEqual(act, exp);

        private ExerciseDataTerminalBuilder(TestData<T, S_IN> data, Function streamF, Function terminalF) {
            this.data = data;
            this.streamF = Objects.requireNonNull(streamF);
            this.terminalF = Objects.requireNonNull(terminalF);
        }

        //

        public ExerciseDataTerminalBuilder<T, U, R, S_IN, S_OUT> expectedResult(R expectedResult) {
            this.refResult = expectedResult;
            return this;
        }

        public ExerciseDataTerminalBuilder<T, U, R, S_IN, S_OUT> equalator(BiConsumer equalityAsserter) {
            resultAsserter = (act, exp, ord, par) -> equalityAsserter.accept(act, exp);
            return this;
        }

        public ExerciseDataTerminalBuilder<T, U, R, S_IN, S_OUT> resultAsserter(ResultAsserter resultAsserter) {
            this.resultAsserter = resultAsserter;
            return this;
        }

        // Build method

        public R exercise() {
            S_OUT out = streamF.apply(data.stream()).sequential();
            AbstractPipeline ap = (AbstractPipeline) out;
            boolean isOrdered = StreamOpFlag.ORDERED.isKnown(ap.getStreamFlags());
            StreamShape shape = ap.getOutputShape();

            EnumSet<TerminalTestScenario> tests = EnumSet.allOf(TerminalTestScenario.class);
            // Sequentially collect the output that will be input to the terminal op
            Node<U> node = ap.evaluateToArrayNode(size -> (U[]) new Object[size]);
            if (refResult == null) {
                // Induce the reference result
                S_OUT source = (S_OUT) createPipeline(shape, node.spliterator(),
                                                      StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SIZED,
                                                      false);

                refResult = (R) TerminalTestScenario.SINGLE_SEQUENTIAL.run(terminalF, source, shape);
                tests.remove(TerminalTestScenario.SINGLE_SEQUENTIAL);
            }

            for (BaseTerminalTestScenario test : tests) {
                S_OUT source;
                if (test.requiresSingleStageSource()) {
                    source = (S_OUT) createPipeline(shape, node.spliterator(),
                                                    StreamOpFlag.IS_ORDERED | StreamOpFlag.IS_SIZED,
                                                    test.requiresParallelSource());
                }
                else {
                    source = streamF.apply(test.requiresParallelSource()
                                           ? data.parallelStream() : data.stream());
                }

                R result = (R) test.run(terminalF, source, shape);

                LambdaTestHelpers.launderAssertion(
                        () -> resultAsserter.assertResult(result, refResult, isOrdered, test.requiresParallelSource()),
                        () -> String.format("%s: %s != %s", test, refResult, result));
            }

            return refResult;
        }

        AbstractPipeline createPipeline(StreamShape shape, Spliterator s, int flags, boolean parallel) {
            switch (shape) {
                case REFERENCE:    return new ReferencePipeline.Head<>(s, flags, parallel);
                case INT_VALUE:    return new IntPipeline.Head(s, flags, parallel);
                case LONG_VALUE:   return new LongPipeline.Head(s, flags, parallel);
                case DOUBLE_VALUE: return new DoublePipeline.Head(s, flags, parallel);
                default: throw new IllegalStateException("Unknown shape: " + shape);
            }
        }
    }

    public <T, R> R exerciseTerminalOps(Collection data, Function, R> m, R expected) {
        TestData.OfRef<T> data1
                = TestData.Factory.ofCollection("Collection of type " + data.getClass().getName(), data);
        return withData(data1).terminal(m).expectedResult(expected).exercise();
    }

    public <T, R, S_IN extends BaseStream R
    exerciseTerminalOps(TestData<T, S_IN> data,
                        Function<S_IN, R> terminalF) {
        return withData(data).terminal(terminalF).exercise();
    }

    public <T, U, R, S_IN extends BaseStream> R
    exerciseTerminalOps(TestData<T, S_IN> data,
                        Function<S_IN, S_OUT> streamF,
                        Function<S_OUT, R> terminalF) {
        return withData(data).terminal(streamF, terminalF).exercise();
    }

    //

    @SuppressWarnings({"rawtypes", "unchecked"})
    private static <T> AbstractPipeline chain(AbstractPipeline upstream, IntermediateTestOp op) {
        return (AbstractPipeline<?, T, ?>) IntermediateTestOp.chain(upstream, op);
    }

    @SuppressWarnings({"rawtypes", "unchecked"})
    private static AbstractPipeline<?, ?, ?> chain(AbstractPipeline pipe, IntermediateTestOp... ops) {
        for (IntermediateTestOp op : ops)
            pipe = chain(pipe, op);
        return pipe;
    }

    @SuppressWarnings("rawtypes")
    private static <T> AbstractPipeline chain(BaseStream pipe, IntermediateTestOp op) {
        return chain((AbstractPipeline) pipe, op);
    }

    @SuppressWarnings("rawtypes")
    public static AbstractPipeline<?, ?, ?> chain(BaseStream pipe, IntermediateTestOp... ops) {
        return chain((AbstractPipeline) pipe, ops);
    }

    // Test data

    static class ShortCircuitOp<T> implements StatelessTestOp {
        private final StreamShape shape;

        ShortCircuitOp(StreamShape shape) {
            this.shape = shape;
        }

        @Override
        public Sink<T> opWrapSink(int flags, boolean parallel, Sink sink) {
            return sink;
        }

        @Override
        public int opGetFlags() {
            return StreamOpFlag.IS_SHORT_CIRCUIT;
        }

        @Override
        public StreamShape outputShape() {
            return shape;
        }

        @Override
        public StreamShape inputShape() {
            return shape;
        }
    }
}