@Test
public void testCompleteJob() throws Exception {

  File inputFile = getTestTempFile("prefs.txt");
  File outputDir = getTestTempDir("output");
  outputDir.delete();
  File similaritiesOutputDir = getTestTempDir("outputSimilarities");
  similaritiesOutputDir.delete();
  File tmpDir = getTestTempDir("tmp");
  writeLines(inputFile,
      "1,1,5",
      "1,2,5",
      "1,3,2",
      "2,1,2",
      "2,3,3",
      "2,4,5",
      "3,2,5",
      "3,4,3",
      "4,1,3",
      "4,4,5");
  RecommenderJob recommenderJob = new RecommenderJob();
  Configuration conf = getConfiguration();
  conf.set("mapred.input.dir", inputFile.getAbsolutePath());
  conf.set("mapred.output.dir", outputDir.getAbsolutePath());
  conf.setBoolean("mapred.output.compress", false);
  recommenderJob.setConf(conf);
  recommenderJob.run(new String[] { "--tempDir", tmpDir.getAbsolutePath(), "--similarityClassname",
     TanimotoCoefficientSimilarity.class.getName(), "--numRecommendations", "4",
      "--outputPathForSimilarityMatrix", similaritiesOutputDir.getAbsolutePath() });
  Map<Long,List<RecommendedItem>> recommendations = readRecommendations(new File(outputDir, "part-r-00000"));
  assertEquals(4, recommendations.size());
  for (Entry<Long,List<RecommendedItem>> entry : recommendations.entrySet()) {
    long userID = entry.getKey();
    List<RecommendedItem> items = entry.getValue();
    assertNotNull(items);
    RecommendedItem item1 = items.get(0);
    if (userID == 1L) {
      assertEquals(1, items.size());
      assertEquals(4L, item1.getItemID());
      assertEquals(4.3, item1.getValue(), 0.05);
    }
    if (userID == 2L) {
      assertEquals(1, items.size());
      assertEquals(2L, item1.getItemID());
      assertEquals(3.3, item1.getValue(), 0.05);
    }
    if (userID == 3L) {
      assertEquals(2, items.size());
      assertEquals(3L, item1.getItemID());
      assertEquals(4.1, item1.getValue(), 0.05);
      RecommendedItem item2 = items.get(1);
      assertEquals(1L, item2.getItemID());
      assertEquals(3.7, item2.getValue(), 0.05);
    }
    if (userID == 4L) {
      assertEquals(2, items.size());
      assertEquals(2L, item1.getItemID());
      assertEquals(4.0, item1.getValue(), 0.05);
      RecommendedItem item2 = items.get(1);
      assertEquals(3L, item2.getItemID());
      assertEquals(3.5, item2.getValue(), 0.05);
    }
  }
  Map<Pair<Long, Long>, Double> similarities = readSimilarities(new File(similaritiesOutputDir, "part-r-00000"));
  assertEquals(6, similarities.size());
  assertEquals(0.25, similarities.get(new Pair<Long, Long>(1L, 2L)), EPSILON);
  assertEquals(0.6666666666666666, similarities.get(new Pair<Long, Long>(1L, 3L)), EPSILON);
  assertEquals(0.5, similarities.get(new Pair<Long, Long>(1L, 4L)), EPSILON);
  assertEquals(0.3333333333333333, similarities.get(new Pair<Long, Long>(2L, 3L)), EPSILON);
  assertEquals(0.25, similarities.get(new Pair<Long, Long>(2L, 4L)), EPSILON);
  assertEquals(0.25, similarities.get(new Pair<Long, Long>(3L, 4L)), EPSILON);
}