@Test
public void testLocalizationInit() throws Exception {

  conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077");
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(new Configuration());
  ContainerExecutor exec = mock(ContainerExecutor.class);
  DeletionService delService = spy(new DeletionService(exec));
  delService.init(conf);
  delService.start();
  List<Path> localDirs = new ArrayList<Path>();
  String[] sDirs = new String[4];
  for (int i = 0; i < 4; ++i) {
    localDirs.add(lfs.makeQualified(new Path(basedir, i + "")));
    sDirs[i] = localDirs.get(i).toString();
  }
  conf.setStrings(YarnConfiguration.NM_LOCAL_DIRS, sDirs);
  LocalDirsHandlerService diskhandler = new LocalDirsHandlerService();
  diskhandler.init(conf);
  ResourceLocalizationService locService =
    spy(new ResourceLocalizationService(dispatcher, exec, delService,
                                        diskhandler));
  doReturn(lfs)
    .when(locService).getLocalFileContext(isA(Configuration.class));
  try {
    dispatcher.start();
    // initialize ResourceLocalizationService
    locService.init(conf);
    final FsPermission defaultPerm = new FsPermission((short)0755);
    // verify directory creation
    for (Path p : localDirs) {
      p = new Path((new URI(p.toString())).getPath());
      Path usercache = new Path(p, ContainerLocalizer.USERCACHE);
      verify(spylfs)
        .mkdir(eq(usercache),
            eq(defaultPerm), eq(true));
      Path publicCache = new Path(p, ContainerLocalizer.FILECACHE);
      verify(spylfs)
        .mkdir(eq(publicCache),
            eq(defaultPerm), eq(true));
      Path nmPriv = new Path(p, ResourceLocalizationService.NM_PRIVATE_DIR);
      verify(spylfs).mkdir(eq(nmPriv),
          eq(ResourceLocalizationService.NM_PRIVATE_PERM), eq(true));
    }
  } finally {
    dispatcher.stop();
    delService.stop();
  }
}