@VisibleForTesting
  boolean verifyAccess() throws IOException {
    // if it is in the public cache, it's trivially OK
    if (resource.getVisibility() == LocalResourceVisibility.PUBLIC) {
      return true;
    }

    final Path remotePath;
    try {
      remotePath = resource.getResource().toPath();
    } catch (URISyntaxException e) {
      throw new IOException("Invalid resource", e);
    }

    // get the file status of the HDFS file
    FileSystem remoteFs = remotePath.getFileSystem(conf);
    FileStatus status = remoteFs.getFileStatus(remotePath);
    // check to see if the file has been modified in any way
    if (status.getModificationTime() != resource.getTimestamp()) {
      LOG.warn("The remote file " + remotePath +
          " has changed since it's localized; will not consider it for upload");
      return false;
    }

    // check for the user ownership
    if (status.getOwner().equals(user)) {
      return true; // the user owns the file
    }
    // check if the file is publicly readable otherwise
    return fileIsPublic(remotePath, remoteFs, status);
  }
@Test
  public void testVerifyAccessPublicResource() throws Exception {
    Configuration conf = new Configuration();
    conf.setBoolean(YarnConfiguration.SHARED_CACHE_ENABLED, true);
    LocalResource resource = mock(LocalResource.class);
    // give public visibility
    when(resource.getVisibility()).thenReturn(LocalResourceVisibility.PUBLIC);
    Path localPath = mock(Path.class);
    when(localPath.getName()).thenReturn("foo.jar");
    String user = "joe";
    SCMUploaderProtocol scmClient = mock(SCMUploaderProtocol.class);
    FileSystem fs = mock(FileSystem.class);
    FileSystem localFs = FileSystem.getLocal(conf);
    SharedCacheUploader spied =
        createSpiedUploader(resource, localPath, user, conf, scmClient, fs,
            localFs);

    assertTrue(spied.verifyAccess());
  }