@Test
public void testMapReduce() throws Exception {

  Path tmpDir = getTestTempDirPath();
  Path mrOutputDir = new Path(tmpDir, "mail-archives-out-mr");
  Configuration configuration = getConfiguration();
  FileSystem fs = FileSystem.get(configuration);
  File expectedInputFile = new File(inputDir.toString());
  String[] args = {
    "-Dhadoop.tmp.dir=" + configuration.get("hadoop.tmp.dir"),
    "--input", expectedInputFile.getAbsolutePath(),
    "--output", mrOutputDir.toString(),
    "--charset", "UTF-8",
    "--keyPrefix", "TEST",
    "--method", "mapreduce",
    "--body", "--subject", "--separator", ""
  };
  // run the application's main method
  SequenceFilesFromMailArchives.main(args);
  // app should create a single SequenceFile named "chunk-0" in the output dir
  FileStatus[] fileStatuses = fs.listStatus(mrOutputDir.suffix("/part-m-00000"));
  assertEquals(1, fileStatuses.length); // only one
  assertEquals("part-m-00000", fileStatuses[0].getPath().getName());
  SequenceFileIterator<Text, Text> iterator =
    new SequenceFileIterator<Text, Text>(mrOutputDir.suffix("/part-m-00000"), true, configuration);
  Assert.assertTrue("First key/value pair not found!", iterator.hasNext());
  Pair<Text, Text> record = iterator.next();
  File parentFileSubSubDir = new File(new File(new File(new File("TEST"), "subdir"), "subsubdir"), "mail-messages-2.gz");
  String expected = record.getFirst().toString();
  if (SystemUtils.IS_OS_WINDOWS) {
    expected = expected.replace("/", "\\");
  }
  Assert.assertEquals(new File(parentFileSubSubDir, testVars[0][0]).toString(), expected);
  Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString());
  Assert.assertTrue("Second key/value pair not found!", iterator.hasNext());
  record = iterator.next();
  expected = record.getFirst().toString();
  if (SystemUtils.IS_OS_WINDOWS) {
    expected = expected.replace("/", "\\");
  }
  Assert.assertEquals(new File(parentFileSubSubDir, testVars[1][0]).toString(), expected);
  Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString());
  // test other file
  File parentFile = new File(new File(new File("TEST"), "subdir"), "mail-messages.gz");
  record = iterator.next();
  expected = record.getFirst().toString();
  if (SystemUtils.IS_OS_WINDOWS) {
    expected = expected.replace("/", "\\");
  }
  Assert.assertEquals(new File(parentFile, testVars[0][0]).toString(), expected);
  Assert.assertEquals(testVars[0][1] + testVars[0][2], record.getSecond().toString());
  Assert.assertTrue("Second key/value pair not found!", iterator.hasNext());
  record = iterator.next();
  expected = record.getFirst().toString();
  if (SystemUtils.IS_OS_WINDOWS) {
    expected = expected.replace("/", "\\");
  }
  Assert.assertEquals(new File(parentFile, testVars[1][0]).toString(), expected);
  Assert.assertEquals(testVars[1][1] + testVars[1][2], record.getSecond().toString());
  Assert.assertFalse("Only four key/value pairs expected!", iterator.hasNext());
}