@Test
public void testMirrorStreamValidator() throws Exception {

  Date date = new Date();
  Calendar cal = Calendar.getInstance();
  cal.setTime(date);
  cal.add(Calendar.MINUTE, 1);
  Date nextDate = cal.getTime();
  cal.add(Calendar.MINUTE, 4);
  Date stopDate = cal.getTime();
  ConduitConfig config = setup("test-mirror-validator-conduit.xml");
  // clean up all root dir before generating test data
  cleanUp(config);
  createMergeData(config, date);
  Set<String> streamsSet = config.getSourceStreams().keySet();
  FileSystem fs = FileSystem.getLocal(new Configuration());
  String auditSrcJar = FileUtil.findContainingJar(
      com.inmobi.messaging.util.AuditUtil.class);
  for (String streamName : streamsSet) {
    for (Cluster cluster : config.getClusters().values()) {
      if (cluster.getMirroredStreams().contains(streamName)) {
        jarsPath = new Path(cluster.getTmpPath(), "jars");
        auditUtilJarDestPath = new Path(jarsPath, "messaging-client-core.jar");
        // Copy AuditUtil src jar to FS
        fs.copyFromLocalFile(new Path(auditSrcJar), auditUtilJarDestPath);
        createMirrorData(config, streamName, cluster, date);
        //check whether given start time is valid
        testStartTimeBeyondRetention(config,streamName, cluster.getName(),date,
            nextDate);
        // it tests missing paths for given a specific period
        testMirrorValidatorVerify(config,streamName, cluster.getName(),date,
            nextDate, false, false);
        // verify : it tests what all are the missing paths
        testMirrorValidatorVerify(config,streamName, cluster.getName(),date,
            stopDate, false, true);
        // fix : throw an exception as there are holes in source
        testMirrorValidatorFixWithHolesInSource(config,streamName,
            cluster.getName(), date, stopDate);
        // fix : It copies all the missing paths to mirror cluster
        testMirrorValidatorFix(config,streamName, cluster.getName(), date,
            stopDate);
        // reverify : should not contain any missing paths after fixing
        testMirrorValidatorVerify(config,streamName, cluster.getName(),date,
            stopDate, true, true);
      }
    }
  }
  cleanUp(config);
}