AffMatchingService implements Serializable { public void matchAffiliations(JavaSparkContext sc, String inputAffPath, String inputOrgPath, String outputPath, String outputReportPath) { checkArguments(sc, inputAffPath, inputOrgPath, outputPath); checkState(); JavaRDD<AffMatchAffiliation> affiliations = affiliationReader.readAffiliations(sc, inputAffPath); JavaRDD<AffMatchOrganization> organizations = organizationReader.readOrganizations(sc, inputOrgPath); JavaRDD<AffMatchAffiliation> normalizedAffiliations = affiliations.map(aff -> affMatchAffiliationNormalizer.normalize(aff)).filter(aff -> (StringUtils.isNotBlank(aff.getOrganizationName()))); JavaRDD<AffMatchOrganization> normalizedOrganizations = organizations.map(org -> affMatchOrganizationNormalizer.normalize(org)).filter(org -> (StringUtils.isNotBlank(org.getName()))); JavaRDD<AffMatchOrganization> enrichedOrganizations = normalizedOrganizations.map(x -> affMatchOrganizationAltNameFiller.fillAlternativeNames(x)); JavaRDD<AffMatchResult> allMatchedAffOrgs = doMatch(sc, normalizedAffiliations, enrichedOrganizations); affMatchResultWriter.write(sc, allMatchedAffOrgs, outputPath, outputReportPath, numberOfEmittedFiles); }  void matchAffiliations(JavaSparkContext sc, String inputAffPath, String inputOrgPath, String outputPath, String outputReportPath); void setOrganizationReader(OrganizationReader organizationReader); void setAffiliationReader(AffiliationReader affiliationReader); void setAffMatchOrganizationAltNameFiller(AffMatchOrganizationAltNameFiller affMatchOrganizationAltNameFiller); void setAffOrgMatchers(List<AffOrgMatcher> affOrgMatchers); void setAffMatchResultWriter(AffMatchResultWriter affMatchResultWriter); void setNumberOfEmittedFiles(int numberOfEmittedFiles);  }
@Test(expected = NullPointerException.class) public void matchAffiliations_sparkContext_null() { affMatchingService.matchAffiliations(null, inputAffPath, inputOrgPath, outputPath, outputReportPath); }
@Test(expected = IllegalArgumentException.class) public void matchAffiliations_inputAffPath_blank() { affMatchingService.matchAffiliations(sc, " ", inputOrgPath, outputPath, outputReportPath); }
@Test(expected = IllegalArgumentException.class) public void matchAffiliations_inputOrgPath_blank() { affMatchingService.matchAffiliations(sc, inputAffPath, " ", outputPath, outputReportPath); }
@Test(expected = IllegalArgumentException.class) public void matchAffiliations_outputPath_blank() { affMatchingService.matchAffiliations(sc, inputAffPath, inputOrgPath, " ", outputReportPath); }
@Test public void matchAffiliations() throws Exception { when(affiliationReader.readAffiliations(sc, inputAffPath)).thenReturn(affiliations); when(organizationReader.readOrganizations(sc, inputOrgPath)).thenReturn(organizations); doReturn(normalizedAffiliations).when(affiliations).map(Mockito.any()); doReturn(normalizedOrganizations).when(organizations).map(Mockito.any()); when(normalizedAffiliations.filter(Mockito.any())).thenReturn(normalizedAndFilteredAffiliations); when(normalizedOrganizations.filter(Mockito.any())).thenReturn(normalizedAndFilteredOrganizations); doReturn(altNamesFilledOrganizations).when(normalizedAndFilteredOrganizations).map(Mockito.any()); doReturn(parallelizedAffMatchResults).when(sc).parallelizePairs(new ArrayList<>()); when(affOrgMatcher1.match(normalizedAndFilteredAffiliations, altNamesFilledOrganizations)).thenReturn(matchedAffOrgs1); doReturn(matchedAffOrgsWithKey1).when(matchedAffOrgs1).keyBy(Mockito.any()); when(parallelizedAffMatchResults.union(matchedAffOrgsWithKey1)).thenReturn(allAffMatchResults1); when(affOrgMatcher2.match(normalizedAndFilteredAffiliations, altNamesFilledOrganizations)).thenReturn(matchedAffOrgs2); doReturn(matchedAffOrgsWithKey2).when(matchedAffOrgs2).keyBy(Mockito.any()); when(allAffMatchResults1.union(matchedAffOrgsWithKey2)).thenReturn(allAffMatchResults2); doReturn(allUniqueAffMatchResults).when(allAffMatchResults2).reduceByKey(Mockito.any()); when(allUniqueAffMatchResults.values()).thenReturn(allUniqueAffMatchResultsValues); affMatchingService.matchAffiliations(sc, inputAffPath, inputOrgPath, outputPath, outputReportPath); verify(affMatchResultWriter).write(sc, allUniqueAffMatchResultsValues, outputPath, outputReportPath, 1); verify(affiliationReader).readAffiliations(sc, inputAffPath); verify(organizationReader).readOrganizations(sc, inputOrgPath); verify(affiliations).map(affNormalizingFunction.capture()); assertAffNormalizingFunction(affNormalizingFunction.getValue()); verify(organizations).map(orgNormalizingFunction.capture()); assertOrgNormalizingFunction(orgNormalizingFunction.getValue()); verify(normalizedAffiliations).filter(affFilterFunction.capture()); assertAffFilterFunction(affFilterFunction.getValue()); verify(normalizedOrganizations).filter(orgFilterFunction.capture()); assertOrgFilterFunction(orgFilterFunction.getValue()); verify(normalizedAndFilteredOrganizations).map(orgFillAltNamesFunction.capture()); assertOrgFillAltNamesFunctionFunction(orgFillAltNamesFunction.getValue()); verify(matchedAffOrgs1).keyBy(matchedAffOrgs1KeyByFunction.capture()); assertMatchedAffOrgsKeyByFunction(matchedAffOrgs1KeyByFunction.getValue()); verify(matchedAffOrgs2).keyBy(matchedAffOrgs2KeyByFunction.capture()); assertMatchedAffOrgsKeyByFunction(matchedAffOrgs2KeyByFunction.getValue()); verify(allAffMatchResults2).reduceByKey(allAffMatchResultsReduceFunction.capture()); assertAllAffMatchResultsReduceFunction(allAffMatchResultsReduceFunction.getValue()); }