List<DataflowPackage> stageClasspathElements(
      Collection<StagedFile> classpathElements, String stagingPath, CreateOptions createOptions) {
    return stageClasspathElements(classpathElements, stagingPath, DEFAULT_SLEEPER, createOptions);
  }
@Test
  public void testPackageUploadWithDirectorySucceeds() throws Exception {
    Pipe pipe = Pipe.open();
    File tmpDirectory = tmpFolder.newFolder("folder");
    tmpFolder.newFolder("folder", "empty_directory");
    tmpFolder.newFolder("folder", "directory");
    makeFileWithContents("folder/file.txt", "This is a test!");
    makeFileWithContents("folder/directory/file.txt", "This is also a test!");

    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());

    defaultPackageUtil.stageClasspathElements(
        ImmutableList.of(makeStagedFile(tmpDirectory.getAbsolutePath())),
        STAGING_PATH,
        createOptions);

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

    List<String> zipEntryNames = new ArrayList<>();
    try (ZipInputStream inputStream = new ZipInputStream(Channels.newInputStream(pipe.source()))) {
      for (ZipEntry entry = inputStream.getNextEntry();
          entry != null;
          entry = inputStream.getNextEntry()) {
        zipEntryNames.add(entry.getName());
      }
    }

    assertThat(
        zipEntryNames, containsInAnyOrder("directory/file.txt", "empty_directory/", "file.txt"));
  }