Running Tests

Setting up Tests with Flinkspector using TestBase or StreamTestBase bases is easy:

class Test extends StreamTestBase {

    @org.junit.Test
    public myTest() {
        DataSet<Integer> dataSet = createTestDataSet(asList(1,2,3))
            .map((MapFunction<Integer,Integer>) (value) -> {return value + 1});

        ExpectedOutput<Integer> expectedOutput = 
            new ExpectedOutput<Integer>().expectAll(asList(2,3,4))

        assertDataSet(dataSet, expectedOutput);
    }

}

Alternatively tests can be defined like this:

@org.junit.Test
    public myTest() {
        DataSet<Integer> dataSet = createTestDataSet(asList(1,2,3))
            .map((MapFunction<Integer,Integer>) (value) -> {return value + 1});

        ExpectedOutput<Integer> expectedOutput = 
            new ExpectedOutput<Integer>().expectAll(asList(2,3,4))

        dataSet.output(createTestOutputFormat(expectedOutput));
}

Flinkspector will execute the test environment automatically.

##Finishing tests early Test will run with a default timeout of 4 seconds. If the timeout has expired the framework checks whether the Matchers have been fulfilled. If not you will get an according exception. Change the timeout with .setTimeout(long interval), interval in milliseconds.

For each .assertStream() or .asserDataSet() you use, you can set a VerifyFinishedTrigger:

assertDataSet(dataSet, matcher, FinishAtCount.of(3));
assertStream(dataStream, matcher, FinishAtMatch.of(is(Tuple2.of("rose",12))));

• FinishAtCount.of(3) will finish the verification when 3 records have been produced by the DataSet.
• FinishAtMatch.of(is(Tuple2.of("rose",12))) finishes the verification when the passed hamcrest matcher returns a positive match.

All triggers work in batch and stream tests.

If all triggers in a test case fired the test run will be stopped prematurely. This gives you the possibility to finish tests early if your data flow is producing more output than you wan't to see or the data flow under test does not terminate regularly if the input closes.

A trigger can also be passed to createTestOutputFormat() or createTestSink():

TestSink<Integer> sink = 
		createTestSink(greaterThan(3), FinishAtMatch.of(is(1)));
                                 
TestOutputFormat<Integer> outputformat = 
		createTestOutputFormat(lessThan(2), FinishAtCount.of(3));