@Test
public void readCitations() throws Exception {

    
    // given
    
    String inputPath = "/this/is/a/path";
    
    
    when(sparkContext.sequenceFile(inputPath, Writable.class, BytesWritable.class)).thenReturn(rawCitations);
    doReturn(docWrappers).when(rawCitations).map(Mockito.any());
    doReturn(docReferences).when(docWrappers).flatMapToPair(Mockito.any());
    
    // execute
    
    JavaPairRDD<String, ReferenceMetadata> citReferences = coansysInputCitationReader.readCitations(sparkContext, inputPath);
    
    
    // assert
    
    assertTrue(docReferences == citReferences);
    
    verify(sparkContext).sequenceFile(inputPath, Writable.class, BytesWritable.class);
    
    verify(rawCitations).map(convertMatchedCitationFunction.capture());
    assertConvertToDocumentWrapperFunction(convertMatchedCitationFunction.getValue());
    
    verify(docWrappers).flatMapToPair(extractReferencesFunction.capture());
    assertExtractReferencesFunction(extractReferencesFunction.getValue());
    
}