  @Test
  public void parseBooks_returnsNgramsCounts() {

    // Arrange
    Pipeline p = TestPipeline.create();
    PCollection<String> input = p.apply(Create.of(testFile));

    // Act
    PCollection<KV<String, Integer>> output = LoadBooks.applyPipelineToParseBooks(input);

    // Assert
    DataflowAssert.that(output)
        .containsInAnyOrder(
            KV.of("despatch when art", 10),
            KV.of("despatch when came", 10),
            KV.of("despatch when published", 12),
            KV.of("despatch where was", 10),
            KV.of("despatch which made", 45),
            // There are two entries for "despatch which addressed".
            // Each entry has a different part of speech for "addressed".
            KV.of("despatch which addressed", 12 + 46),
            KV.of("despatch which admitted", 13),
            KV.of("despatch which allow", 14),
            KV.of("despatch which announced", 50),
            KV.of("despatch which answer", 32));
  }