### Question:
PathToPathMetadata implements Function<Path, PathMetadata> { @Override public PathMetadata apply(@Nonnull Path location) { try { FileSystem fs = location.getFileSystem(conf); FileStatus fileStatus = fs.getFileStatus(location); FileChecksum checksum = null; if (fileStatus.isFile()) { checksum = fs.getFileChecksum(location); } long modificationTime = 0; List<PathMetadata> childPathDescriptors = new ArrayList<>(); if (fileStatus.isDirectory()) { FileStatus[] childStatuses = fs.listStatus(location); for (FileStatus childStatus : childStatuses) { childPathDescriptors.add(apply(childStatus.getPath())); } } else { modificationTime = fileStatus.getModificationTime(); } return new PathMetadata(location, modificationTime, checksum, childPathDescriptors); } catch (IOException e) { throw new CircusTrainException("Unable to compute digest for location " + location.toString(), e); } } PathToPathMetadata(Configuration conf); @Override PathMetadata apply(@Nonnull Path location);  }### Answer:
@Test public void file() throws Exception { when(fileStatus.isFile()).thenReturn(true); when(fileStatus.isDirectory()).thenReturn(false); when(fs.getFileChecksum(path)).thenReturn(fileChecksum); PathMetadata metadata = function.apply(path); assertThat(metadata.getLocation(), is(DIR_PATH)); assertThat(metadata.getLastModifiedTimestamp(), is(LAST_MODIFIED)); assertThat(metadata.getChecksumAlgorithmName(), is(CHECKSUM_ALGORITHM)); assertThat(metadata.getChecksumLength(), is(CHECKSUM_LENGTH)); assertThat(metadata.getChecksum(), is(CHECKSUM_BYTES)); assertThat(metadata.getChildrenMetadata().isEmpty(), is(true)); verify(fs, never()).listStatus(any(Path.class)); }
@Test public void directory() throws Exception { when(fileStatus.isFile()).thenReturn(false); when(fileStatus.isDirectory()).thenReturn(true); Path childPath = mock(Path.class); when(childPath.toUri()).thenReturn(new URI(FILE_PATH)); when(childPath.getFileSystem(any(Configuration.class))).thenReturn(fs); FileStatus childStatus = mock(FileStatus.class); when(childStatus.getPath()).thenReturn(childPath); when(childStatus.getModificationTime()).thenReturn(LAST_MODIFIED_CHILD); when(childStatus.isFile()).thenReturn(true); when(childStatus.isDirectory()).thenReturn(false); when(fs.getFileStatus(childPath)).thenReturn(childStatus); when(fs.getFileChecksum(childPath)).thenReturn(fileChecksum); FileStatus[] childStatuses = new FileStatus[] { childStatus }; when(fs.listStatus(path)).thenReturn(childStatuses); PathMetadata metadata = function.apply(path); assertThat(metadata.getLocation(), is(DIR_PATH)); assertThat(metadata.getLastModifiedTimestamp(), is(0L)); assertThat(metadata.getChecksumAlgorithmName(), is(nullValue())); assertThat(metadata.getChecksumLength(), is(0)); assertThat(metadata.getChecksum(), is(nullValue())); assertThat(metadata.getChildrenMetadata().size(), is(1)); verify(fs, times(1)).listStatus(path); verify(fs, never()).getFileChecksum(path); PathMetadata childMetadata = metadata.getChildrenMetadata().get(0); assertThat(childMetadata.getLocation(), is(FILE_PATH)); assertThat(childMetadata.getLastModifiedTimestamp(), is(LAST_MODIFIED_CHILD)); assertThat(childMetadata.getChecksumAlgorithmName(), is(CHECKSUM_ALGORITHM)); assertThat(childMetadata.getChecksumLength(), is(CHECKSUM_LENGTH)); assertThat(childMetadata.getChecksum(), is(CHECKSUM_BYTES)); assertThat(childMetadata.getChildrenMetadata().size(), is(0)); verify(fs, never()).listStatus(childPath); verify(fs, times(1)).getFileChecksum(childPath); }