@Test
public void listStatus() throws IOException {

    // Create temp files
    final File file1 = tempFolder.newFile("file1");
    Assert.assertTrue(file1.setLastModified(currentTimeMillis));
    final File folder1 = tempFolder.newFolder("folder1");
    final File file2 = tempFolder.newFile("folder1/file2");
    Assert.assertTrue(file2.setLastModified(currentTimeMillis));
    tempFolder.newFolder("folder1", "folder2");
    final File file3 = tempFolder.newFile("folder1/folder2/file3");
    Assert.assertTrue(file3.setLastModified(currentTimeMillis));
    // Test listing files
    final Job job = Job.getInstance(new Configuration(false));
    HighWaterMarkInputFormat.setInputPaths(job, file1.getAbsolutePath() + "," + folder1.getAbsolutePath());
    final HighWaterMarkInputFormat inputFormat = new MockHighWaterMarkInputFormat();
    final List<FileStatus> files = inputFormat.listStatus(job);
    Collections.sort(files, new FileStatusComparator());
    Assert.assertEquals(new Path(file1.toURI()), files.get(0).getPath());
    Assert.assertEquals(new Path(file2.toURI()), files.get(1).getPath());
    Assert.assertEquals(2, files.size());
    Assert.assertEquals(currentTimeMillis, inputFormat.getLastHighWaterMark());
    // Verify files are not listed again
    Assert.assertEquals(0, inputFormat.listStatus(job).size());
}