@Test
public void testCopyMapperImplMethod() throws Exception {

  ConduitConfigParser parser = new ConduitConfigParser(
      "test-lss-conduit-s3n.xml");
  Set<String> streamsToProcess = new HashSet<String>();
  ConduitConfig config = parser.getConfig();
  streamsToProcess.addAll(config.getSourceStreams().keySet());
  Set<String> clustersToProcess = new HashSet<String>();
  Set<TestLocalStreamService> services = new HashSet<TestLocalStreamService>();
  for (SourceStream sStream : config.getSourceStreams().values()) {
    for (String cluster : sStream.getSourceClusters()) {
      clustersToProcess.add(cluster);
    }
  }
  for (String clusterName : clustersToProcess) {
    Cluster cluster = config.getClusters().get(clusterName);
    cluster.getHadoopConf().set("mapred.job.tracker",
        super.CreateJobConf().get("mapred.job.tracker"));
    TestLocalStreamService service = new TestLocalStreamService(config,
        cluster, null, new NullCheckPointProvider(), streamsToProcess);
    services.add(service);
  }
  for (TestLocalStreamService service : services) {
    Assert.assertEquals(service.getMapperClass(), S3NCopyMapper.class);
  }
  Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_MKDIR,"test1").getValue().longValue() , 0);
  Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_CHECKPOINT,"test1").getValue().longValue() , 0);
  Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.RETRY_RENAME,"test1").getValue().longValue() , 0);
  Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.EMPTYDIR_CREATE,"test1").getValue().longValue() , 0);
  Assert.assertEquals(ConduitMetrics.<SlidingTimeWindowGauge>getMetric("LocalStreamService",AbstractService.FILES_COPIED_COUNT,"test1").getValue().longValue() , 0);
}