### Question:
HdfsDeployer { public void deploy() throws Exception { FileSystem fs = context.getFileSystem(); final String hdfsDirLocalPath = String.format(LOCAL_HDFS_PATTERN, context.getDeployDir()); final File hdfsDirLocal = new File(hdfsDirLocalPath); if (!hdfsDirLocal.exists()) { throw new IllegalStateException(hdfsDirLocalPath + " not found local FS"); } undeploy(); Path dst = getDestinationHdfsPath(); fs.mkdirs(dst); String[] childFiles = hdfsDirLocal.list(); for (String child : childFiles) { fs.copyFromLocalFile(new Path(hdfsDirLocalPath, child), dst); } } HdfsDeployer(CelosCiContext context); void undeploy(); void deploy();  }### Answer:
@Test(expected = IllegalStateException.class) public void testDeployThrowsExceptionNoDir() throws Exception { CelosCiContext context = mock(CelosCiContext.class); HdfsDeployer deployer = new HdfsDeployer(context); doReturn(LocalFileSystem.get(new Configuration())).when(context).getFileSystem(); doReturn(new File("nodir" + UUID.randomUUID())).when(context).getDeployDir(); deployer.deploy(); }
@Test public void testDeployRemoteDirNotExist() throws Exception { CelosCiContext context = mock(CelosCiContext.class); HdfsDeployer deployer = new HdfsDeployer(context); File localFolder = tempDir.newFolder(); File remoteHdfsFolder = tempDir.newFolder(); File localHdfsFolder = new File(localFolder, "hdfs"); localHdfsFolder.mkdirs(); new File(localHdfsFolder, "file1").createNewFile(); new File(localHdfsFolder, "file2").createNewFile(); doReturn(LocalFileSystem.get(new Configuration())).when(context).getFileSystem(); doReturn(localFolder).when(context).getDeployDir(); doReturn(remoteHdfsFolder.getAbsolutePath()).when(context).getHdfsPrefix(); doReturn("workflow").when(context).getWorkflowName(); doReturn("/some/hdfs/root").when(context).getHdfsRoot(); File targetDir = new File(remoteHdfsFolder, "some/hdfs/root/workflow"); deployer.deploy(); Set<String> resultSet = getFilesWithoutCrc(targetDir); Assert.assertEquals(Sets.newHashSet("file2", "file1"), resultSet); }
@Test public void testDeployRemoteDirExist() throws Exception { CelosCiContext context = mock(CelosCiContext.class); HdfsDeployer deployer = new HdfsDeployer(context); File localFolder = tempDir.newFolder(); File remoteHdfsFolder = tempDir.newFolder(); File localHdfsFolder = new File(localFolder, "hdfs"); localHdfsFolder.mkdirs(); new File(localHdfsFolder, "file1").createNewFile(); new File(localHdfsFolder, "file2").createNewFile(); doReturn(LocalFileSystem.get(new Configuration())).when(context).getFileSystem(); doReturn(localFolder).when(context).getDeployDir(); doReturn(remoteHdfsFolder.getAbsolutePath()).when(context).getHdfsPrefix(); doReturn("workflow").when(context).getWorkflowName(); doReturn("/some/hdfs/root").when(context).getHdfsRoot(); File targetDir = new File(remoteHdfsFolder, "some/hdfs/root/workflow"); targetDir.mkdirs(); new File(targetDir, "file3").createNewFile(); new File(targetDir, "file4").createNewFile(); Assert.assertEquals(Sets.newHashSet("file3", "file4"), getFilesWithoutCrc(targetDir)); deployer.deploy(); Assert.assertEquals(Sets.newHashSet("file2", "file1"), getFilesWithoutCrc(targetDir)); }