@Test
public void testGetSplits() throws Exception {

  final DistCpContext context = new DistCpContext(getOptions());
  Configuration configuration = new Configuration();
  configuration.set("mapred.map.tasks",
                    String.valueOf(context.getMaxMaps()));
  CopyListing.getCopyListing(configuration, CREDENTIALS, context)
      .buildListing(new Path(cluster.getFileSystem().getUri().toString()
          +"/tmp/testDynInputFormat/fileList.seq"), context);
  JobContext jobContext = new JobContextImpl(configuration, new JobID());
  DynamicInputFormat<Text, CopyListingFileStatus> inputFormat =
      new DynamicInputFormat<Text, CopyListingFileStatus>();
  List<InputSplit> splits = inputFormat.getSplits(jobContext);
  int nFiles = 0;
  int taskId = 0;
  for (InputSplit split : splits) {
    StubContext stubContext = new StubContext(jobContext.getConfiguration(),
                                              null, taskId);
    final TaskAttemptContext taskAttemptContext
       = stubContext.getContext();
    RecordReader<Text, CopyListingFileStatus> recordReader =
        inputFormat.createRecordReader(split, taskAttemptContext);
    stubContext.setReader(recordReader);
    recordReader.initialize(splits.get(0), taskAttemptContext);
    float previousProgressValue = 0f;
    while (recordReader.nextKeyValue()) {
      CopyListingFileStatus fileStatus = recordReader.getCurrentValue();
      String source = fileStatus.getPath().toString();
      System.out.println(source);
      Assert.assertTrue(expectedFilePaths.contains(source));
      final float progress = recordReader.getProgress();
      Assert.assertTrue(progress >= previousProgressValue);
      Assert.assertTrue(progress >= 0.0f);
      Assert.assertTrue(progress <= 1.0f);
      previousProgressValue = progress;
      ++nFiles;
    }
    Assert.assertTrue(recordReader.getProgress() == 1.0f);
    ++taskId;
  }
  Assert.assertEquals(expectedFilePaths.size(), nFiles);
}