examples/iot-nirvana/pipeline/src/test/java/com/google/cloud/demo/iot/nirvana/pipeline/FilterMessagesTest.java
/**
   * Test {@link FilterMessages.processElement} with a mix of valid and invalid messages, whose
   * respective identifiers corresponds to coordinates of an actual cities.
   */
Test  FilterMessages.processElement with a mix of valid and invalid messages, whose
respective identifiers corresponds to coordinates of an actual cities.
Test FilterMessages . processElement with a mix of valid and invalid messages whose respective identifiers corresponds to coordinates of an actual cities .
@Test
  @Category(NeedsRunner.class)
  public void processElement_mixedValidInvalidMessage() throws Exception {
    List<String> inputMessages =
        Arrays.asList(INVALID_MESSAGE, VALID_MESSAGE_TIRANA_STR, VALID_MESSAGE_ORANJESTAD_STR);
    List<Message> expectedValidResult =
        Arrays.asList(VALID_MESSAGE_TIRANA, VALID_MESSAGE_ORANJESTAD);
    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).containsInAnyOrder(expectedValidResult);
    PAssert.that(invalidTemperatures).containsInAnyOrder(INVALID_MESSAGE_RESULT);
    pipeline.run();
  }
@ Test @ Category ( NeedsRunner . class ) public void processElement_mixedValidInvalidMessage ( ) throws Exception { List < String > inputMessages = Arrays . asList ( INVALID_MESSAGE , VALID_MESSAGE_TIRANA_STR , VALID_MESSAGE_ORANJESTAD_STR ) ; List < Message > expectedValidResult = Arrays . asList ( VALID_MESSAGE_TIRANA , VALID_MESSAGE_ORANJESTAD ) ; 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 ) . containsInAnyOrder ( expectedValidResult ) ; PAssert . that ( invalidTemperatures ) . containsInAnyOrder ( INVALID_MESSAGE_RESULT ) ; pipeline . run ( ) ; }
Test {@link FilterMessages.processElement} with a mix of valid and invalid messages, whose
 respective identifiers corresponds to coordinates of an actual cities.
Test { @link FilterMessages . processElement } with a mix of valid and invalid messages whose respective identifiers corresponds to coordinates of an actual cities .
// Create test data // Create pipeline graph // Run pipeline and evaluate results