unittest.2.java

public class StopWordRemoveFnTest {

    @Rule
    public final transient TestPipeline pipeline = TestPipeline.create();

    @Test
    public void testDoFn_TestPipeline() throws Exception {
        PCollection<KV<Empty, String>> input = pipeline.apply(Create.of(
            KV.of(Empty.EMPTY, "be"), KV.of(Empty.EMPTY, "is"), KV.of(Empty.EMPTY, "night"), KV.of(Empty.EMPTY, "dream")
        ).withCoder(KvCoder.of(AvroCoder.of(Empty.class), StringUtf8Coder.of())));

        PCollection<KV<Empty, String>> output = input.apply(ParDo.of(new StopWordRemoveFn<>()));

        PAssert.that(output)
            .containsInAnyOrder(
                KV.of(Empty.EMPTY, "night"),
                KV.of(Empty.EMPTY, "dream")
            );

        // run pipeline!
        pipeline.run();
    }
}