examples/iot-nirvana/pipeline/src/test/java/com/google/cloud/demo/iot/nirvana/pipeline/FilterMessagesTest.java
/**
   * Test {@link FilterMessages.processElement} with a malformed message, which does not comply to
   * JSON formalism.
   */
Test  FilterMessages.processElement with a malformed message, which does not comply to
JSON formalism.
Test FilterMessages . processElement with a malformed message which does not comply to JSON formalism .
@Test
  @Category(NeedsRunner.class)
  public void processElement_malformedMessage() throws Exception {
    List<String> inputMessages = Arrays.asList(MALFORMED_MESSAGE);
    PCollectionTuple filteredMessages =
        pipeline
            .apply("Create", Create.of(inputMessages))
            .apply(
                "FilterMessages",
                ParDo.of(
                        FilterMessages.newBuilder()
                            .setCities(cities)
                            .setErrorTag(ERROR_TAG)
                            .build())
                    .withSideInputs(cities)
                    .withOutputTags(GOOD_TAG, TupleTagList.of(ERROR_TAG)));
    PCollection<Message> validTemperatures = filteredMessages.get(GOOD_TAG);
    PCollection<String> invalidTemperatures = filteredMessages.get(ERROR_TAG);
    PAssert.that(validTemperatures).empty();
    PAssert.that(invalidTemperatures).containsInAnyOrder(MALFORMED_MESSAGE_RESULT);
    pipeline.run();
  }
@ Test @ Category ( NeedsRunner . class ) public void processElement_malformedMessage ( ) throws Exception { List < String > inputMessages = Arrays . asList ( MALFORMED_MESSAGE ) ; PCollectionTuple filteredMessages = pipeline . apply ( "Create" , Create . of ( inputMessages ) ) . apply ( "FilterMessages" , ParDo . of ( FilterMessages . newBuilder ( ) . setCities ( cities ) . setErrorTag ( ERROR_TAG ) . build ( ) ) . withSideInputs ( cities ) . withOutputTags ( GOOD_TAG , TupleTagList . of ( ERROR_TAG ) ) ) ; PCollection < Message > validTemperatures = filteredMessages . get ( GOOD_TAG ) ; PCollection < String > invalidTemperatures = filteredMessages . get ( ERROR_TAG ) ; PAssert . that ( validTemperatures ) . empty ( ) ; PAssert . that ( invalidTemperatures ) . containsInAnyOrder ( MALFORMED_MESSAGE_RESULT ) ; pipeline . run ( ) ; }
Test {@link FilterMessages.processElement} with a malformed message, which does not comply to
 JSON formalism.
Test { @link FilterMessages . processElement } with a malformed message which does not comply to JSON formalism .
// Create test data // Create pipeline graph // Run pipeline and evaluate results