List<DataflowPackage> stageClasspathElements(
      Collection<StagedFile> classpathElements, String stagingPath, CreateOptions createOptions) {
    return stageClasspathElements(classpathElements, stagingPath, DEFAULT_SLEEPER, createOptions);
  }
@Test
  public void testPackageUploadWithEmptyDirectorySucceeds() throws Exception {
    Pipe pipe = Pipe.open();
    File tmpDirectory = tmpFolder.newFolder("folder");

    when(mockGcsUtil.getObjects(anyListOf(GcsPath.class)))
        .thenReturn(
            ImmutableList.of(
                StorageObjectOrIOException.create(new FileNotFoundException("some/path"))));
    when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)))
        .thenReturn(pipe.sink());

    List<DataflowPackage> targets =
        defaultPackageUtil.stageClasspathElements(
            ImmutableList.of(makeStagedFile(tmpDirectory.getAbsolutePath())),
            STAGING_PATH,
            createOptions);
    DataflowPackage target = Iterables.getOnlyElement(targets);

    verify(mockGcsUtil).getObjects(anyListOf(GcsPath.class));
    verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
    verifyNoMoreInteractions(mockGcsUtil);

    assertThat(target.getName(), endsWith(".jar"));
    assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
    try (ZipInputStream zipInputStream =
        new ZipInputStream(Channels.newInputStream(pipe.source()))) {
      assertNull(zipInputStream.getNextEntry());
    }
  }