@Test
public void matchCitations() {

    
    // given
    
    when(inputCitationReader.readCitations(sparkContext, "/input/cit/path")).thenReturn(citations);
    when(inputCitationConverter.convertCitations(citations)).thenReturn(convertedCitations);
    when(convertedCitations.partitions()).thenReturn(Lists.newArrayList(16));
    when(convertedCitations.partitionBy(any())).thenReturn(repartitionedCitations);
    
    when(inputDocumentReader.readDocuments(sparkContext, "/input/doc/path")).thenReturn(documents);
    when(inputDocumentConverter.convertDocuments(documents)).thenReturn(convertedDocuments);
    when(convertedDocuments.partitions()).thenReturn(Lists.newArrayList(12));
    when(convertedDocuments.partitionBy(any())).thenReturn(repartitionedDocuments);
    
    when(coreCitationMatchingService.matchCitations(repartitionedCitations, repartitionedDocuments)).thenReturn(matched);
    when(outputConverter.convertMatchedCitations(matched)).thenReturn(convertedMatched);
    
    
    // execute
    
    citationMatchingService.matchCitations(sparkContext, "/input/cit/path", "/input/doc/path", "/output/path");
    
    
    // assert
    
    verify(inputCitationReader).readCitations(sparkContext, "/input/cit/path");
    verify(inputCitationConverter).convertCitations(citations);
    verify(convertedCitations).partitionBy(citationsPartitioner.capture());
    assertPartitioner(citationsPartitioner.getValue(), 5);
    
    
    verify(inputDocumentReader).readDocuments(sparkContext, "/input/doc/path");
    verify(inputDocumentConverter).convertDocuments(documents);
    verify(convertedDocuments).partitionBy(documentsPartitioner.capture());
    assertPartitioner(documentsPartitioner.getValue(), 5);
    
    verify(coreCitationMatchingService).matchCitations(repartitionedCitations, repartitionedDocuments);
    
    verify(outputConverter).convertMatchedCitations(matched);
    verify(outputWriter).writeMatchedCitations(convertedMatched, "/output/path");
}