Running Tests

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

class Test extends DataStreamTestBase {

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

        ExpectedRecords<Integer> expected = 
            new ExpectedRecords<Integer>().expectAll(asList(2,3,4))

        assertDataSet(dataSet, expected);
    }

}

Alternatively tests can be set up like this:

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

     dataSet.output(createTestOutputFormat(expected));
}

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 .assertDataSet() 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));