@Test
public void testBuild() throws Exception {

  final KijiMapReduceJob job = KijiMapReduceJobBuilder.create()
      .withConf(mConf)
      .withInput(MapReduceJobInputs.newTextMapReduceJobInput(new Path("/path/to/my/input")))
      .withMapper(MyMapper.class)
      .withReducer(MyReducer.class)
      .withOutput(MapReduceJobOutputs.newTextMapReduceJobOutput(
          new Path("/path/to/my/output"), 16))
      .build();
  final Job hadoopJob = job.getHadoopJob();
  assertEquals(TextInputFormat.class, hadoopJob.getInputFormatClass());
  assertEquals(MyMapper.class, hadoopJob.getMapperClass());
  assertEquals(MyReducer.class, hadoopJob.getReducerClass());
  assertEquals(16, hadoopJob.getNumReduceTasks());
  assertEquals(TextOutputFormat.class, hadoopJob.getOutputFormatClass());
  // KeyValueStore-specific checks here.
  Configuration confOut = hadoopJob.getConfiguration();
  assertEquals(2, confOut.getInt(KeyValueStoreConfigSerializer.CONF_KEY_VALUE_STORE_COUNT, 0));
  assertEquals(EmptyKeyValueStore.class.getName(),
      confOut.get(KeyValueStoreConfiguration.KEY_VALUE_STORE_NAMESPACE + "0."
      + KeyValueStoreConfigSerializer.CONF_CLASS));
  assertEquals("mapperMap",
      confOut.get(KeyValueStoreConfiguration.KEY_VALUE_STORE_NAMESPACE + "0."
      + KeyValueStoreConfigSerializer.CONF_NAME));
  assertEquals(EmptyKeyValueStore.class.getName(),
      confOut.get(KeyValueStoreConfiguration.KEY_VALUE_STORE_NAMESPACE + "1."
      + KeyValueStoreConfigSerializer.CONF_CLASS));
  assertEquals("reducerMap",
      confOut.get(KeyValueStoreConfiguration.KEY_VALUE_STORE_NAMESPACE + "1."
      + KeyValueStoreConfigSerializer.CONF_NAME));
}