AffOrgMatchComputer implements Serializable { public JavaRDD<AffMatchResult> computeMatches(JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> joinedAffOrgs) { checkNotNull(joinedAffOrgs); checkState(CollectionUtils.isNotEmpty(affOrgMatchVoters), "no AffOrgMatchVoter has been set"); JavaRDD<AffMatchResult> affMatchResults = joinedAffOrgs.map(kv->new AffMatchResult(kv._1(), kv._2(), 0)); for (AffOrgMatchVoter voter : affOrgMatchVoters) { affMatchResults = affMatchResults.map(affOrgMatch -> affOrgMatchStrengthRecalculator.recalculateMatchStrength(affOrgMatch, voter)); } affMatchResults = affMatchResults.filter(affOrgMatch -> affOrgMatch.getMatchStrength() > 0); return affMatchResults; }  JavaRDD<AffMatchResult> computeMatches(JavaRDD<Tuple2<AffMatchAffiliation, AffMatchOrganization>> joinedAffOrgs); void setAffOrgMatchVoters(List<AffOrgMatchVoter> affOrgMatchVoters);  }
@Test(expected = NullPointerException.class) public void computeMatches_null() { affOrgMatchComputer.computeMatches(null); }
@Test public void computeMatches() throws Exception { doReturn(affMatchResults).when(joinedAffOrgs).map(Mockito.any()); doReturn(recalcAffMatchResults1).when(affMatchResults).map(Mockito.any()); doReturn(recalcAffMatchResults2).when(recalcAffMatchResults1).map(Mockito.any()); doReturn(filteredAffMatchResults).when(recalcAffMatchResults2).filter(Mockito.any()); JavaRDD<AffMatchResult> retAffMatchResults = affOrgMatchComputer.computeMatches(joinedAffOrgs); assertNotNull(retAffMatchResults); assertTrue(filteredAffMatchResults == retAffMatchResults); verify(joinedAffOrgs).map(mapToMatchResultFunction.capture()); assertMapToMatchResultFunction(mapToMatchResultFunction.getValue()); verify(affMatchResults).map(recalcMatchStrengthFunction.capture()); assertRecalcMatchStrengthFunction(recalcMatchStrengthFunction.getValue(), affOrgMatchVoter1); verify(recalcAffMatchResults1).map(recalcMatchStrengthFunction.capture()); assertRecalcMatchStrengthFunction(recalcMatchStrengthFunction.getValue(), affOrgMatchVoter2); verify(recalcAffMatchResults2).filter(filterAffMatchResultFunction.capture()); assertFilterRecalcAffMatchResultFunction(filterAffMatchResultFunction.getValue()); }