IisOrganizationReader implements Serializable, OrganizationReader { @Override public JavaRDD<AffMatchOrganization> readOrganizations(JavaSparkContext sc, String inputPath) { Preconditions.checkNotNull(sc); Preconditions.checkArgument(StringUtils.isNotBlank(inputPath)); JavaRDD<Organization> sourceOrganizations = sparkAvroLoader.loadJavaRDD(sc, inputPath, Organization.class); JavaRDD<AffMatchOrganization> organizations = sourceOrganizations.map(srcOrg -> organizationConverter.convert(srcOrg)); return organizations; }  @Override JavaRDD<AffMatchOrganization> readOrganizations(JavaSparkContext sc, String inputPath);  }
@Test(expected = NullPointerException.class) public void readOrganizations_sparkContext_null() { reader.readOrganizations(null, "/aaa"); }
@Test(expected = IllegalArgumentException.class) public void readOrganizations_inputPath_blank() { reader.readOrganizations(sparkContext, " "); }
@Test public void readOrganizations() throws Exception { String inputPath = "/data/organizations"; when(sparkAvroLoader.loadJavaRDD(sparkContext, inputPath, Organization.class)).thenReturn(inputOrganizations); doReturn(affMatchOrganizations).when(inputOrganizations).map(any()); JavaRDD<AffMatchOrganization> retAffMatchOrganizations = reader.readOrganizations(sparkContext, inputPath); assertTrue(affMatchOrganizations == retAffMatchOrganizations); verify(inputOrganizations).map(convertFunction.capture()); assertConvertFunction(convertFunction.getValue()); }