@Test
public void testUploadFile() throws Exception{

  final File sourceFile = new File(
      getClass().getClassLoader().getResource("sample_flow_01.zip").getFile());
  final String sourceHash = new String(Hex.encodeHex(HashUtils.MD5.getHashBytes(sourceFile)));
  final File tmpUploadDir = TEMP_DIR.newFolder("upload_test_dir");
  final File destFile = new File(tmpUploadDir, "tmpfile1");
  final FileContext destFileContext = FileContext.getLocalFSFileContext();
  // A new destination file should be created after this, with same content as the source file.
  HdfsStorage.uploadLocalFile(new Path(sourceFile.getPath()), new Path(destFile.getPath()),
      destFileContext);
  final String destHash = new String(Hex.encodeHex(HashUtils.MD5.getHashBytes(destFile)));
  Assert.assertEquals(sourceHash, destHash);
  // Verify that copying over an existing destination file works as well.
  final byte[] newFileContent = "12345".getBytes(UTF_8);
  final File newSourceFile = new File(tmpUploadDir, "tmpsource1");
  FileUtils.writeByteArrayToFile(newSourceFile, newFileContent, false);
  // The destination file already exists from previous invocation of uploadLocalFile.
  HdfsStorage.uploadLocalFile(new Path(newSourceFile.getPath()), new Path(destFile.getPath()),
      destFileContext);
  final byte[] destFileContent = FileUtils.readFileToByteArray(destFile);
  FileUtils.deleteDirectory(tmpUploadDir);
  Assert.assertArrayEquals(newFileContent, destFileContent);
}