@Test
public void testUpload() throws Exception {

  ZooKeeperManager zkManager = mock(ZooKeeperManager.class);
  Job job = mock(Job.class);
  Configuration conf = new Configuration();
  TerrapinUploaderOptions options = new TerrapinUploaderOptions();
  options.terrapinFileSet = FILE_SET;
  when(zkManager.getClusterInfo()).thenReturn(new ClusterInfo(NAME_NODE, REPLICA_FACTOR));
  doNothing().when(zkManager).lockFileSet(anyString(), any(FileSetInfo.class));
  doNothing().when(zkManager).unlockFileSet(anyString());
  when(job.waitForCompletion(anyBoolean())).then(new Answer<Object>() {
    @Override
    public Object answer(InvocationOnMock invocationOnMock) throws Throwable {
      Thread.sleep(1000);
      return true;
    }
  });
  when(job.getNumReduceTasks()).thenReturn(NUM_PART);
  when(job.getConfiguration()).thenReturn(conf);
  doNothing().when(job).setOutputFormatClass(Matchers.<Class<? extends OutputFormat>>any());
  doNothing().when(job).setOutputKeyClass(Matchers.<Class<?>>any());
  doNothing().when(job).setOutputValueClass(Matchers.<Class<?>>any());
  TestHadoopJobLoader uploader = new TestHadoopJobLoader(zkManager, options, job);
  assertTrue(uploader.waitForCompletion());
  assertEquals(Constants.MAPRED_MAP_MAX_ATTEMPTS,
      Integer.parseInt(conf.get("mapred.map.max.attempts")));
  assertEquals(Constants.CHECKSUM_BYTES,
      Integer.parseInt(conf.get("io.bytes.per.checksum")));
  assertEquals(Constants.DEFAULT_MAX_SHARD_SIZE_BYTES,
      Long.parseLong(conf.get("dfs.block.size")));
  assertEquals(REPLICA_FACTOR, Integer.parseInt(conf.get("dfs.replication")));
  verify(zkManager).getClusterInfo();
  verify(zkManager).lockFileSet(eq(FILE_SET), any(FileSetInfo.class));
  verify(zkManager).unlockFileSet(eq(FILE_SET));
  verify(job).waitForCompletion(eq(true));
  verify(job).setOutputFormatClass(eq(HFileOutputFormat.class));
  verify(job).setOutputValueClass(eq(BytesWritable.class));
  verify(job).setOutputKeyClass(eq(BytesWritable.class));
}