content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.beam.runners.dataflow.util;
  
  import static org.hamcrest.MatcherAssert.assertThat;
  import static org.hamcrest.Matchers.endsWith;
  import static org.hamcrest.Matchers.equalTo;
  import static org.hamcrest.Matchers.instanceOf;
  import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder;
  import static org.junit.Assert.assertEquals;
  import static org.junit.Assert.assertNotEquals;
  import static org.junit.Assert.assertNull;
  import static org.junit.Assert.assertTrue;
  import static org.junit.Assert.fail;
  import static org.mockito.ArgumentMatchers.any;
  import static org.mockito.ArgumentMatchers.anyList;
  import static org.mockito.Mockito.times;
  import static org.mockito.Mockito.verify;
  import static org.mockito.Mockito.verifyNoMoreInteractions;
  import static org.mockito.Mockito.when;
  
  import com.google.api.client.googleapis.json.GoogleJsonError.ErrorInfo;
  import com.google.api.client.googleapis.json.GoogleJsonResponseException;
  import com.google.api.client.http.HttpRequest;
  import com.google.api.client.http.HttpResponse;
  import com.google.api.client.http.HttpStatusCodes;
  import com.google.api.client.http.HttpTransport;
  import com.google.api.client.http.LowLevelHttpRequest;
  import com.google.api.client.json.GenericJson;
  import com.google.api.client.json.Json;
  import com.google.api.client.json.JsonFactory;
  import com.google.api.client.json.gson.GsonFactory;
  import com.google.api.client.testing.http.HttpTesting;
  import com.google.api.client.testing.http.MockHttpTransport;
  import com.google.api.client.testing.http.MockLowLevelHttpRequest;
  import com.google.api.client.testing.http.MockLowLevelHttpResponse;
  import com.google.api.services.dataflow.model.DataflowPackage;
  import com.google.api.services.storage.model.StorageObject;
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.io.IOException;
  import java.math.BigInteger;
  import java.nio.channels.Channels;
  import java.nio.channels.Pipe;
  import java.nio.charset.StandardCharsets;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.List;
  import java.util.zip.ZipEntry;
  import java.util.zip.ZipInputStream;
  import org.apache.beam.runners.dataflow.util.PackageUtil.PackageAttributes;
  import org.apache.beam.runners.dataflow.util.PackageUtil.StagedFile;
  import org.apache.beam.sdk.extensions.gcp.options.GcsOptions;
  import org.apache.beam.sdk.extensions.gcp.util.GcsUtil;
  import org.apache.beam.sdk.extensions.gcp.util.GcsUtil.StorageObjectOrIOException;
  import org.apache.beam.sdk.extensions.gcp.util.gcsfs.GcsPath;
  import org.apache.beam.sdk.io.FileSystems;
  import org.apache.beam.sdk.io.fs.CreateOptions;
  import org.apache.beam.sdk.io.fs.CreateOptions.StandardCreateOptions;
  import org.apache.beam.sdk.io.fs.ResolveOptions.StandardResolveOptions;
  import org.apache.beam.sdk.options.PipelineOptionsFactory;
  import org.apache.beam.sdk.testing.ExpectedLogs;
  import org.apache.beam.sdk.testing.RegexMatcher;
  import org.apache.beam.sdk.util.FastNanoClockAndSleeper;
  import org.apache.beam.sdk.util.MimeTypes;
  import org.apache.beam.sdk.util.ZipFiles;
  import org.apache.beam.sdk.util.construction.Environments;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.hash.HashCode;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.Files;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.io.LineReader;
  import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.util.concurrent.MoreExecutors;
  import org.checkerframework.checker.nullness.qual.Nullable;
  import org.hamcrest.Matchers;
  import org.junit.After;
  import org.junit.Before;
  import org.junit.Rule;
  import org.junit.Test;
  import org.junit.rules.TemporaryFolder;
  import org.junit.runner.RunWith;
  import org.junit.runners.JUnit4;
  import org.mockito.Mock;
  import org.mockito.MockitoAnnotations;
  
  /** Tests for {@link PackageUtil}. */
  @RunWith(JUnit4.class)
  public class PackageUtilTest {
    @Rule public ExpectedLogs logged = ExpectedLogs.none(PackageUtil.class);
    @Rule public TemporaryFolder tmpFolder = new TemporaryFolder();
    @Rule public FastNanoClockAndSleeper fastNanoClockAndSleeper = new FastNanoClockAndSleeper();
  
    @Mock GcsUtil mockGcsUtil;
  
    private CreateOptions createOptions;
    private PackageUtil defaultPackageUtil;
  
    @Before
    public void setUp() {
      MockitoAnnotations.initMocks(this);
  
      GcsOptions pipelineOptions = PipelineOptionsFactory.as(GcsOptions.class);
      pipelineOptions.setGcsUtil(mockGcsUtil);
      FileSystems.setDefaultPipelineOptions(pipelineOptions);
      createOptions = StandardCreateOptions.builder().setMimeType(MimeTypes.BINARY).build();
      defaultPackageUtil = PackageUtil.withDefaultThreadPool();
    }
  
    @After
    public void teardown() {
      defaultPackageUtil.close();
    }
  
    private File makeFileWithContents(String name, String contents) throws Exception {
      File tmpFile = tmpFolder.newFile(name);
      Files.asCharSink(tmpFile, StandardCharsets.UTF_8).write(contents);
      assertTrue(tmpFile.setLastModified(0)); // required for determinism with directories
      return tmpFile;
    }
  
    static final GcsPath STAGING_GCS_PATH = GcsPath.fromComponents("somebucket", "base/path/");
    static final String STAGING_PATH = STAGING_GCS_PATH.toString();
  
    private static PackageAttributes makePackageAttributes(
        File file, @Nullable String overridePackageName) throws IOException {
      StagedFile stagedFile = makeStagedFile(file.getPath());
      PackageAttributes attributes =
          PackageUtil.PackageAttributes.forFileToStage(
              stagedFile.getSource(),
              stagedFile.getSha256(),
              stagedFile.getDestination(),
              STAGING_PATH);
      if (overridePackageName != null) {
        attributes = attributes.withPackageName(overridePackageName);
      }
      return attributes;
    }
  
    private static StagedFile makeStagedFile(String source) throws IOException {
      return makeStagedFile(source, null);
    }
  
    private static StagedFile makeStagedFile(String source, String destName) throws IOException {
      File file = new File(source);
      File sourceFile;
      HashCode hashCode;
      if (file.exists()) {
        sourceFile = file.isDirectory() ? zipDirectory(file) : file;
        hashCode = Files.asByteSource(sourceFile).hash(Hashing.sha256());
      } else {
        sourceFile = file;
        hashCode = Hashing.sha256().hashBytes(new byte[] {});
      }
      String destination =
          destName == null ? Environments.createStagingFileName(file, hashCode) : destName;
      return StagedFile.of(sourceFile.getPath(), hashCode.toString(), destination);
    }
  
    private static File zipDirectory(File directory) throws IOException {
      File tempDir = Files.createTempDir();
      File sourceFile = new File(tempDir, directory.getName() + ".jar");
      ZipFiles.zipDirectory(directory, sourceFile);
      return sourceFile;
    }
  
    @Test
    public void testFileWithExtensionPackageNamingAndSize() throws Exception {
      String contents = "This is a test!";
      File tmpFile = makeFileWithContents("file.txt", contents);
      PackageAttributes attr = makePackageAttributes(tmpFile, null);
      DataflowPackage target = attr.getDestination();
  
      assertThat(target.getName(), endsWith(".txt"));
      assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
      assertThat(attr.getSize(), equalTo((long) contents.length()));
    }
  
    @Test
    public void testPackageNamingWithFileNoExtension() throws Exception {
      File tmpFile = makeFileWithContents("file", "This is a test!");
      DataflowPackage target = makePackageAttributes(tmpFile, null).getDestination();
  
      assertThat(Files.getFileExtension(target.getName()), equalTo(""));
      assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
    }
  
    @Test
    public void testPackageNamingWithDirectory() throws Exception {
      File tmpDirectory = tmpFolder.newFolder("folder");
      DataflowPackage target = makePackageAttributes(tmpDirectory, null).getDestination();
  
      assertThat(target.getName(), endsWith(".jar"));
      assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
    }
  
    @Test
    public void testPackageNamingWithFilesHavingSameContentsAndSameNames() throws Exception {
      File tmpDirectory1 = tmpFolder.newFolder("folder1", "folderA");
      makeFileWithContents("folder1/folderA/sameName", "This is a test!");
      DataflowPackage target1 = makePackageAttributes(tmpDirectory1, null).getDestination();
  
      File tmpDirectory2 = tmpFolder.newFolder("folder2", "folderA");
      makeFileWithContents("folder2/folderA/sameName", "This is a test!");
      DataflowPackage target2 = makePackageAttributes(tmpDirectory2, null).getDestination();
  
      assertEquals(target1.getName(), target2.getName());
      assertEquals(target1.getLocation(), target2.getLocation());
    }
  
    @Test
    public void testPackageNamingWithFilesHavingSameContentsButDifferentNames() throws Exception {
      File tmpDirectory1 = tmpFolder.newFolder("folder1", "folderA");
      makeFileWithContents("folder1/folderA/uniqueName1", "This is a test!");
      DataflowPackage target1 = makePackageAttributes(tmpDirectory1, null).getDestination();
  
      File tmpDirectory2 = tmpFolder.newFolder("folder2", "folderA");
      makeFileWithContents("folder2/folderA/uniqueName2", "This is a test!");
      DataflowPackage target2 = makePackageAttributes(tmpDirectory2, null).getDestination();
  
      assertNotEquals(target1.getName(), target2.getName());
      assertNotEquals(target1.getLocation(), target2.getLocation());
    }
  
    @Test
    public void testPackageNamingWithDirectoriesHavingSameContentsButDifferentNames()
        throws Exception {
      File tmpDirectory1 = tmpFolder.newFolder("folder1", "folderA");
      tmpFolder.newFolder("folder1", "folderA", "uniqueName1");
      DataflowPackage target1 = makePackageAttributes(tmpDirectory1, null).getDestination();
  
      File tmpDirectory2 = tmpFolder.newFolder("folder2", "folderA");
      tmpFolder.newFolder("folder2", "folderA", "uniqueName2");
      DataflowPackage target2 = makePackageAttributes(tmpDirectory2, null).getDestination();
  
      assertNotEquals(target1.getName(), target2.getName());
      assertNotEquals(target1.getLocation(), target2.getLocation());
    }
  
    @Test
    public void testPackageUploadWithLargeClasspathLogsWarning() throws Exception {
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(
                      createStorageObject(STAGING_PATH, tmpFile.length()))));
  
      List<StagedFile> classpathElements = Lists.newLinkedList();
      for (int i = 0; i < 1005; ++i) {
        String eltName = "element" + i;
        classpathElements.add(makeStagedFile(tmpFile.getAbsolutePath(), eltName));
      }
  
      defaultPackageUtil.stageClasspathElements(classpathElements, STAGING_PATH, createOptions);
      logged.verifyWarn("Your classpath contains 1005 elements, which Google Cloud Dataflow");
    }
  
    @Test
    public void testPackageUploadWithFileSucceeds() throws Exception {
      Pipe pipe = Pipe.open();
      String contents = "This is a test!";
      File tmpFile = makeFileWithContents("file.txt", contents);
      when(mockGcsUtil.getObjects(anyList()))
          .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(tmpFile.getAbsolutePath())),
              STAGING_PATH,
              createOptions);
      DataflowPackage target = Iterables.getOnlyElement(targets);
  
      verify(mockGcsUtil).getObjects(anyList());
      verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
      verifyNoMoreInteractions(mockGcsUtil);
  
      assertThat(target.getName(), endsWith(".txt"));
      assertThat(target.getLocation(), equalTo(STAGING_PATH + target.getName()));
      assertThat(
          new LineReader(Channels.newReader(pipe.source(), StandardCharsets.UTF_8.name())).readLine(),
          equalTo(contents));
    }
  
    @Test
    public void testStagingPreservesClasspath() throws Exception {
      File smallFile = makeFileWithContents("small.txt", "small");
      File largeFile = makeFileWithContents("large.log", "large contents");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(new FileNotFoundException("some/path"))));
  
      when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)))
          .thenAnswer(invocation -> Pipe.open().sink());
  
      List<DataflowPackage> targets =
          defaultPackageUtil.stageClasspathElements(
              ImmutableList.of(
                  makeStagedFile(smallFile.getAbsolutePath()),
                  makeStagedFile(largeFile.getAbsolutePath())),
              STAGING_PATH,
              createOptions);
      // Verify that the packages are returned small, then large, matching input order even though
      // the large file would be uploaded first.
      assertThat(targets.get(0).getName(), endsWith(".txt"));
      assertThat(targets.get(1).getName(), endsWith(".log"));
    }
  
    @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(anyList()))
          .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(anyList());
      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"));
    }
  
    @Test
    public void testPackageUploadWithEmptyDirectorySucceeds() throws Exception {
      Pipe pipe = Pipe.open();
      File tmpDirectory = tmpFolder.newFolder("folder");
  
      when(mockGcsUtil.getObjects(anyList()))
          .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(anyList());
      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());
      }
    }
  
    @Test(expected = RuntimeException.class)
    public void testPackageUploadFailsWhenIOExceptionThrown() throws Exception {
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(new FileNotFoundException("some/path"))));
      when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)))
          .thenThrow(new IOException("Fake Exception: Upload error"));
  
      try (PackageUtil directPackageUtil =
          PackageUtil.withExecutorService(MoreExecutors.newDirectExecutorService())) {
        directPackageUtil.stageClasspathElements(
            ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())),
            STAGING_PATH,
            fastNanoClockAndSleeper::sleep,
            createOptions);
      } finally {
        verify(mockGcsUtil, times(5)).getObjects(anyList());
        verify(mockGcsUtil, times(5)).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
        verifyNoMoreInteractions(mockGcsUtil);
      }
    }
  
    @Test
    public void testPackageUploadFailsWithPermissionsErrorGivesDetailedMessage() throws Exception {
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(new FileNotFoundException("some/path"))));
      when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)))
          .thenThrow(
              new IOException(
                  "Failed to write to GCS path " + STAGING_PATH,
                  googleJsonResponseException(
                      HttpStatusCodes.STATUS_CODE_FORBIDDEN, "Permission denied", "Test message")));
  
      try (PackageUtil directPackageUtil =
          PackageUtil.withExecutorService(MoreExecutors.newDirectExecutorService())) {
        directPackageUtil.stageClasspathElements(
            ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())),
            STAGING_PATH,
            fastNanoClockAndSleeper::sleep,
            createOptions);
        fail("Expected RuntimeException");
      } catch (RuntimeException e) {
        assertThat(
            "Expected RuntimeException wrapping IOException.",
            e.getCause(),
            instanceOf(RuntimeException.class));
        assertThat(
            "Expected IOException containing detailed message.",
            e.getCause().getCause(),
            instanceOf(IOException.class));
        assertThat(
            e.getCause().getCause().getMessage(),
            Matchers.allOf(
                Matchers.containsString("Uploaded failed due to permissions error"),
                Matchers.containsString(
                    "Stale credentials can be resolved by executing 'gcloud auth application-default "
                        + "login'")));
      } finally {
        verify(mockGcsUtil).getObjects(anyList());
        verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
        verifyNoMoreInteractions(mockGcsUtil);
      }
    }
  
    @Test
    public void testPackageUploadEventuallySucceeds() throws Exception {
      Pipe pipe = Pipe.open();
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(new FileNotFoundException("some/path"))));
      when(mockGcsUtil.create(any(GcsPath.class), any(GcsUtil.CreateOptions.class)))
          .thenThrow(new IOException("Fake Exception: 410 Gone")) // First attempt fails
          .thenThrow(
              new IOException("Fake Exception: 412 Precondition Failed")) // Second attempt fails
          .thenReturn(pipe.sink()); // Third attempt succeeds
  
      try (PackageUtil directPackageUtil =
          PackageUtil.withExecutorService(MoreExecutors.newDirectExecutorService())) {
        directPackageUtil.stageClasspathElements(
            ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())),
            STAGING_PATH,
            fastNanoClockAndSleeper::sleep,
            createOptions);
      } finally {
        verify(mockGcsUtil, times(3)).getObjects(anyList());
        verify(mockGcsUtil, times(3)).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
        verifyNoMoreInteractions(mockGcsUtil);
      }
    }
  
    @Test
    public void testPackageUploadIsSkippedWhenFileAlreadyExists() throws Exception {
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      when(mockGcsUtil.getObjects(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(
                      createStorageObject(STAGING_PATH, tmpFile.length()))));
  
      defaultPackageUtil.stageClasspathElements(
          ImmutableList.of(makeStagedFile(tmpFile.getAbsolutePath())), STAGING_PATH, createOptions);
  
      verify(mockGcsUtil).getObjects(anyList());
      verifyNoMoreInteractions(mockGcsUtil);
    }
  
    @Test
    public void testPackageUploadIsNotSkippedWhenSizesAreDifferent() 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(anyList()))
          .thenReturn(
              ImmutableList.of(
                  StorageObjectOrIOException.create(
                      createStorageObject(STAGING_PATH, Long.MAX_VALUE))));
      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(anyList());
      verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
      verifyNoMoreInteractions(mockGcsUtil);
    }
  
    @Test
    public void testPackageUploadWithExplicitPackageName() throws Exception {
      Pipe pipe = Pipe.open();
      File tmpFile = makeFileWithContents("file.txt", "This is a test!");
      final String overriddenName = "alias.txt";
  
      when(mockGcsUtil.getObjects(anyList()))
          .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(tmpFile.getAbsolutePath(), overriddenName)),
              STAGING_PATH,
              createOptions);
      DataflowPackage target = Iterables.getOnlyElement(targets);
  
      verify(mockGcsUtil).getObjects(anyList());
      verify(mockGcsUtil).create(any(GcsPath.class), any(GcsUtil.CreateOptions.class));
      verifyNoMoreInteractions(mockGcsUtil);
  
      assertThat(target.getName(), equalTo(overriddenName));
      assertThat(target.getLocation(), RegexMatcher.matches(STAGING_PATH + "alias.txt"));
    }
  
    @Test
    public void testPackageUploadIsSkippedWithNonExistentResource() throws Exception {
      String nonExistentFile =
          FileSystems.matchNewResource(tmpFolder.getRoot().getPath(), true)
              .resolve("non-existent-file", StandardResolveOptions.RESOLVE_FILE)
              .toString();
      assertEquals(
          Collections.EMPTY_LIST,
          defaultPackageUtil.stageClasspathElements(
              ImmutableList.of(makeStagedFile(nonExistentFile)), STAGING_PATH, createOptions));
    }
  
    /** Builds a fake GoogleJsonResponseException for testing API error handling. */
    private static GoogleJsonResponseException googleJsonResponseException(
        final int status, final String reason, final String message) throws IOException {
      final JsonFactory jsonFactory = new GsonFactory();
      HttpTransport transport =
          new MockHttpTransport() {
            @Override
            public LowLevelHttpRequest buildRequest(String method, String url) throws IOException {
              ErrorInfo errorInfo = new ErrorInfo();
              errorInfo.setReason(reason);
              errorInfo.setMessage(message);
              errorInfo.setFactory(jsonFactory);
              GenericJson error = new GenericJson();
              error.set("code", status);
              error.set("errors", Arrays.asList(errorInfo));
              error.setFactory(jsonFactory);
              GenericJson errorResponse = new GenericJson();
              errorResponse.set("error", error);
              errorResponse.setFactory(jsonFactory);
              return new MockLowLevelHttpRequest()
                  .setResponse(
                      new MockLowLevelHttpResponse()
                          .setContent(errorResponse.toPrettyString())
                          .setContentType(Json.MEDIA_TYPE)
                          .setStatusCode(status));
            }
          };
      HttpRequest request =
          transport.createRequestFactory().buildGetRequest(HttpTesting.SIMPLE_GENERIC_URL);
      request.setThrowExceptionOnExecuteError(false);
      HttpResponse response = request.execute();
      return GoogleJsonResponseException.from(jsonFactory, response);
    }
  
    private StorageObject createStorageObject(String gcsFilename, long fileSize) {
      GcsPath gcsPath = GcsPath.fromUri(gcsFilename);
      return new StorageObject()
          .setBucket(gcsPath.getBucket())
          .setName(gcsPath.getObject())
          .setSize(BigInteger.valueOf(fileSize));
    }
  }
  