@Test
public void attachDocuments() {

    
    
    // given
    
    JavaPairRDD<String, String> matchedCitations = sparkContext.parallelizePairs(ImmutableList.of(
            new Tuple2<>(citation1.id(), document1.id()),
            new Tuple2<>(citation1.id(), document2.id()),
            new Tuple2<>(citation4.id(), document2.id()),
            new Tuple2<>(citation4.id(), document5.id())));
    
    JavaPairRDD<String, MatchableEntity> documents = sparkContext.parallelizePairs(generateIdWithEntityTuples(
            Lists.newArrayList(document1, document2, document3, document4, document5)));
    
    
    
    // execute
    
    JavaPairRDD<String, MatchableEntity> actualCitIdDocPairs = documentAttacher.attachDocuments(matchedCitations, documents);
    
    
    // assert
    List<Tuple2<String, MatchableEntity>> expectedCitIdDocPairs = ImmutableList.of(
            new Tuple2<>(citation1.id(), document1),
            new Tuple2<>(citation1.id(), document2),
            new Tuple2<>(citation4.id(), document2),
            new Tuple2<>(citation4.id(), document5));
    
    assertDocAttachedMatchedCitations(actualCitIdDocPairs.collect(), expectedCitIdDocPairs);
    
}