Add Unit Test for OSS, OBS and GCS

dora/underfs/gcs/src/main/java/alluxio/underfs/gcs/GCSUnderFileSystem.java

@@ -230,7 +230,7 @@ protected ObjectListingChunk getObjectListingChunk(String key, boolean recursive
   }
 
   // Get next chunk of listing result.
-  private StorageObjectsChunk getObjectListingChunk(String key, String delimiter,
+  protected StorageObjectsChunk getObjectListingChunk(String key, String delimiter,
       String priorLastKey) {
     StorageObjectsChunk res;
     try {

dora/underfs/gcs/src/main/java/alluxio/underfs/gcs/GCSUnderFileSystemFactory.java

@@ -44,27 +44,41 @@ public GCSUnderFileSystemFactory() {}
 
   @Override
   public UnderFileSystem create(String path, UnderFileSystemConfiguration conf) {
-    Preconditions.checkNotNull(path, "path");
+    Preconditions.checkNotNull(path, "Unable to create UnderFileSystem instance:"
+        + " URI path should not be null");
 
-    if (conf.getInt(PropertyKey.UNDERFS_GCS_VERSION) == GCS_VERSION_TWO) {
-      try {
-        return GCSV2UnderFileSystem.createInstance(new AlluxioURI(path), conf);
-      } catch (IOException e) {
-        LOG.error("Failed to create GCSV2UnderFileSystem.", e);
-        throw Throwables.propagate(e);
-      }
-    } else {
-      try {
-        return GCSUnderFileSystem.createInstance(new AlluxioURI(path), conf);
-      } catch (ServiceException e) {
-        LOG.error("Failed to create GCSUnderFileSystem.", e);
-        throw Throwables.propagate(e);
+    if (checkGCSCredentials(conf)) {
+      if (conf.getInt(PropertyKey.UNDERFS_GCS_VERSION) == GCS_VERSION_TWO) {
+        try {
+          return GCSV2UnderFileSystem.createInstance(new AlluxioURI(path), conf);
+        } catch (IOException e) {
+          LOG.error("Failed to create GCSV2UnderFileSystem.", e);
+          throw Throwables.propagate(e);
+        }
+      } else {
+        try {
+          return GCSUnderFileSystem.createInstance(new AlluxioURI(path), conf);
+        } catch (ServiceException e) {
+          LOG.error("Failed to create GCSUnderFileSystem.", e);
+          throw Throwables.propagate(e);
+        }
       }
     }
+    String err = "GCS credentials or version not available, cannot create GCS Under File System.";
+    throw Throwables.propagate(new IOException(err));
   }
 
   @Override
   public boolean supportsPath(String path) {
     return path != null && path.startsWith(Constants.HEADER_GCS);
   }
+
+  /**
+   * @param conf optional configuration object for the UFS
+   * @return true if access, secret and endpoint keys are present, false otherwise
+   */
+  private boolean checkGCSCredentials(UnderFileSystemConfiguration conf) {
+    return conf.isSet(PropertyKey.GCS_ACCESS_KEY)
+        && conf.isSet(PropertyKey.GCS_SECRET_KEY);
+  }
 }

dora/underfs/gcs/src/test/java/alluxio/underfs/gcs/GCSOutputStreamTest.java

@@ -0,0 +1,205 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.underfs.gcs;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+
+import alluxio.conf.AlluxioConfiguration;
+import alluxio.conf.Configuration;
+import alluxio.conf.PropertyKey;
+
+import org.jets3t.service.ServiceException;
+import org.jets3t.service.impl.rest.httpclient.GoogleStorageService;
+import org.jets3t.service.model.GSObject;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.Mockito;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedOutputStream;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.security.DigestOutputStream;
+
+/**
+ * Unit tests for the {@link GCSOutputStream}.
+ */
+@RunWith(PowerMockRunner.class)
+public class GCSOutputStreamTest {
+  private GoogleStorageService mClient;
+  private File mFile;
+  private BufferedOutputStream mLocalOutputStream;
+  private static AlluxioConfiguration sConf = Configuration.global();
+  private GSObject mObject;
+  private String mMd5Hash = "md5 hash";
+
+  /**
+   * The exception expected to be thrown.
+   */
+  @Rule
+  public final ExpectedException mThrown = ExpectedException.none();
+
+  /**
+   * Sets the properties and configuration before each test runs.
+   */
+  @Before
+  public void before() throws Exception {
+    mClient = Mockito.mock(GoogleStorageService.class);
+    mObject = Mockito.mock(GSObject.class);
+    Mockito.when(mObject.getMd5HashAsBase64()).thenReturn(mMd5Hash);
+    Mockito.when(mClient.putObject(Mockito.anyString(), Mockito.any(GSObject.class)))
+        .thenReturn(mObject);
+    mFile = Mockito.mock(File.class);
+    mLocalOutputStream = Mockito.mock(BufferedOutputStream.class);
+  }
+
+  /**
+   * Tests to ensure IOException is thrown if {@link FileOutputStream}() throws an IOException.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testConstructor() throws Exception {
+    PowerMockito.whenNew(File.class).withArguments(Mockito.anyString()).thenReturn(mFile);
+    String errorMessage = "protocol doesn't support output";
+    PowerMockito.whenNew(FileOutputStream.class).withArguments(mFile)
+        .thenThrow(new IOException(errorMessage));
+    mThrown.expect(IOException.class);
+    mThrown.expectMessage(errorMessage);
+    new GCSOutputStream("testBucketName", "testKey", mClient,
+        sConf.getList(PropertyKey.TMP_DIRS)).close();
+  }
+
+  /**
+   * Tests to ensure {@link GCSOutputStream#write(int)} calls {@link OutputStream#write(int)}.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testWrite1() throws Exception {
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(DigestOutputStream.class)).thenReturn(mLocalOutputStream);
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(FileOutputStream.class)).thenReturn(mLocalOutputStream);
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient,
+        sConf.getList(PropertyKey.TMP_DIRS));
+    stream.write(1);
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+    Mockito.verify(mLocalOutputStream).write(1);
+  }
+
+  /**
+   * Tests to ensure {@link GCSOutputStream#write(byte[], int, int)} calls
+   * {@link OutputStream#write(byte[], int, int)} .
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testWrite2() throws Exception {
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(DigestOutputStream.class)).thenReturn(mLocalOutputStream);
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(FileOutputStream.class)).thenReturn(mLocalOutputStream);
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient,
+        sConf.getList(PropertyKey.TMP_DIRS));
+    byte[] b = new byte[1];
+    stream.write(b, 0, 1);
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+    Mockito.verify(mLocalOutputStream).write(b, 0, 1);
+  }
+
+  /**
+   * Tests to ensure {@link GCSOutputStream#write(byte[])} calls {@link OutputStream#write(byte[])}.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testWrite3() throws Exception {
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(DigestOutputStream.class)).thenReturn(mLocalOutputStream);
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(FileOutputStream.class)).thenReturn(mLocalOutputStream);
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient, sConf
+        .getList(PropertyKey.TMP_DIRS));
+    byte[] b = new byte[1];
+    stream.write(b);
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+    Mockito.verify(mLocalOutputStream).write(b, 0, 1);
+  }
+
+  /**
+   * Tests to ensure IOException is thrown if
+   * {@link GoogleStorageService#putObject(String, GSObject)} throws an
+   * ServiceException.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testCloseError() throws Exception {
+    String errorMessage = "Invoke the createEmptyObject method error.";
+    BufferedInputStream inputStream = PowerMockito.mock(BufferedInputStream.class);
+    PowerMockito.whenNew(BufferedInputStream.class)
+        .withArguments(any(FileInputStream.class)).thenReturn(inputStream);
+    PowerMockito
+        .when(mClient.putObject(Mockito.anyString(), Mockito.any(GSObject.class)))
+        .thenThrow(new ServiceException(errorMessage));
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient, sConf
+        .getList(PropertyKey.TMP_DIRS));
+    mThrown.expect(IOException.class);
+    mThrown.expectMessage(errorMessage);
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+  }
+
+  /**
+   * Tests to ensure {@link File#delete()} is called when close the stream.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testCloseSuccess() throws Exception {
+    PowerMockito.whenNew(File.class).withArguments(Mockito.anyString()).thenReturn(mFile);
+    FileOutputStream outputStream = PowerMockito.mock(FileOutputStream.class);
+    PowerMockito.whenNew(FileOutputStream.class).withArguments(mFile).thenReturn(outputStream);
+    FileInputStream inputStream = PowerMockito.mock(FileInputStream.class);
+    PowerMockito.whenNew(FileInputStream.class).withArguments(mFile).thenReturn(inputStream);
+
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient, sConf
+        .getList(PropertyKey.TMP_DIRS));
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+    Mockito.verify(mFile).delete();
+  }
+
+  /**
+   * Tests to ensure {@link GCSOutputStream#flush()} calls {@link OutputStream#flush()}.
+   */
+  @Test
+  @PrepareForTest(GCSOutputStream.class)
+  public void testFlush() throws Exception {
+    PowerMockito.whenNew(BufferedOutputStream.class)
+        .withArguments(any(DigestOutputStream.class)).thenReturn(mLocalOutputStream);
+    GCSOutputStream stream = new GCSOutputStream("testBucketName", "testKey", mClient, sConf
+        .getList(PropertyKey.TMP_DIRS));
+    stream.flush();
+    stream.close();
+    assertEquals(mMd5Hash, stream.getContentHash().get());
+    Mockito.verify(mLocalOutputStream).flush();
+  }
+}

dora/underfs/gcs/src/test/java/alluxio/underfs/gcs/GCSPositionReaderTest.java

@@ -0,0 +1,85 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.underfs.gcs;
+
+import org.jets3t.service.ServiceException;
+import org.jets3t.service.impl.rest.httpclient.GoogleStorageService;
+import org.jets3t.service.model.GSObject;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+
+import java.io.BufferedInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+
+public class GCSPositionReaderTest {
+  /**
+   * The GCS Position Reader.
+   */
+  private GCSPositionReader mPositionReader;
+  /**
+   * The COS Client.
+   */
+  private GoogleStorageService mClient;
+  /**
+   * The Bucket Name.
+   */
+  private final String mBucketName = "bucket";
+  /**
+   * The Path (or the Key).
+   */
+  private final String mPath = "path";
+  /**
+   * The File Length.
+   */
+  private final long mFileLength = 100L;
+
+  @Before
+  public void before() throws Exception {
+    mClient = Mockito.mock(GoogleStorageService.class);
+    mPositionReader = new GCSPositionReader(mClient, mBucketName, mPath, mFileLength);
+  }
+
+  /**
+   * Test case for {@link GCSPositionReader#openObjectInputStream(long, int)}.
+   */
+  @Test
+  public void openObjectInputStream() throws Exception {
+    GSObject object = Mockito.mock(GSObject.class);
+    BufferedInputStream objectInputStream = Mockito.mock(BufferedInputStream.class);
+    Mockito.when(mClient.getObject(Mockito.anyString(), Mockito.anyString(),
+        Mockito.any(), Mockito.any(), Mockito.any(), Mockito.any(),
+        Mockito.any(), Mockito.any())).thenReturn(object);
+    Mockito.when(object.getDataInputStream()).thenReturn(objectInputStream);
+
+    // test successful open object input stream
+    long position = 0L;
+    int bytesToRead = 10;
+    InputStream inputStream = mPositionReader.openObjectInputStream(position, bytesToRead);
+    Assert.assertTrue(inputStream instanceof BufferedInputStream);
+
+    // test open object input stream with exception
+    Mockito.when(mClient.getObject(ArgumentMatchers.anyString(), ArgumentMatchers.anyString()))
+        .thenThrow(ServiceException.class);
+    try {
+      mPositionReader.openObjectInputStream(position, bytesToRead);
+    } catch (Exception e) {
+      Assert.assertTrue(e instanceof IOException);
+      String errorMessage = String
+          .format("Failed to get object: %s bucket: %s", mPath, mBucketName);
+      Assert.assertEquals(errorMessage, e.getMessage());
+    }
+  }
+}