You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by jw...@apache.org on 2017/02/13 03:54:35 UTC
[2/4] nifi git commit: NIFI-3449: Adding Google Cloud Storage Bundle
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/GCPIntegrationTests.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/GCPIntegrationTests.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/GCPIntegrationTests.java
new file mode 100644
index 0000000..d259e81
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/GCPIntegrationTests.java
@@ -0,0 +1,23 @@
+/*
+ * 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.nifi.processors.gcp;
+
+/**
+ * Dummy interface used for annotating Integration Tests.
+ */
+public interface GCPIntegrationTests {
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialsFactoryTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialsFactoryTest.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialsFactoryTest.java
new file mode 100644
index 0000000..290ec83
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/CredentialsFactoryTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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.nifi.processors.gcp.credentials.factory;
+
+import com.google.auth.oauth2.ComputeEngineCredentials;
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Modifier;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Tests of the validation and credentials provider capabilities of CredentialsFactory.
+ */
+public class CredentialsFactoryTest {
+
+ @Test
+ public void testCredentialPropertyDescriptorClassCannotBeInvoked() throws Exception {
+ Constructor constructor = CredentialPropertyDescriptors.class.getDeclaredConstructor();
+ assertTrue("Constructor of CredentialPropertyDescriptors should be private", Modifier.isPrivate(constructor.getModifiers()));
+ constructor.setAccessible(true);
+ constructor.newInstance();
+ }
+
+ @Test
+ public void testImplicitApplicationDefaultCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.assertValid();
+
+ Map<PropertyDescriptor, String> properties = runner.getProcessContext().getProperties();
+ final CredentialsFactory factory = new CredentialsFactory();
+ final GoogleCredentials credentials = factory.getGoogleCredentials(properties);
+
+ assertNotNull(credentials);
+ }
+
+ @Test
+ public void testExplicitApplicationDefaultCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.USE_APPLICATION_DEFAULT_CREDENTIALS, "true");
+ runner.assertValid();
+
+ Map<PropertyDescriptor, String> properties = runner.getProcessContext().getProperties();
+ final CredentialsFactory factory = new CredentialsFactory();
+ final GoogleCredentials credentials = factory.getGoogleCredentials(properties);
+
+ assertNotNull(credentials);
+ }
+
+ @Test
+ public void testExplicitApplicationDefaultCredentialsExclusive() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.USE_APPLICATION_DEFAULT_CREDENTIALS, "true");
+ runner.setProperty(CredentialPropertyDescriptors.USE_COMPUTE_ENGINE_CREDENTIALS, "true");
+ runner.assertNotValid();
+ }
+
+ @Test
+ public void testJsonFileCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE,
+ "src/test/resources/mock-gcp-service-account.json");
+ runner.assertValid();
+
+ Map<PropertyDescriptor, String> properties = runner.getProcessContext().getProperties();
+ final CredentialsFactory factory = new CredentialsFactory();
+ final GoogleCredentials credentials = factory.getGoogleCredentials(properties);
+
+ assertNotNull(credentials);
+ assertEquals("credentials class should be equal", ServiceAccountCredentials.class,
+ credentials.getClass());
+ }
+
+
+ @Test
+ public void testBadJsonFileCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE,
+ "src/test/resources/bad-mock-gcp-service-account.json");
+ runner.assertNotValid();
+ }
+
+ @Test
+ public void testJsonStringCredentials() throws Exception {
+ final String jsonRead = new String(
+ Files.readAllBytes(Paths.get("src/test/resources/mock-gcp-service-account.json"))
+ );
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON,
+ jsonRead);
+ runner.assertValid();
+
+ Map<PropertyDescriptor, String> properties = runner.getProcessContext().getProperties();
+ final CredentialsFactory factory = new CredentialsFactory();
+ final GoogleCredentials credentials = factory.getGoogleCredentials(properties);
+
+ assertNotNull(credentials);
+ assertEquals("credentials class should be equal", ServiceAccountCredentials.class,
+ credentials.getClass());
+ }
+
+ @Test
+ public void testComputeEngineCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsFactoryProcessor.class);
+ runner.setProperty(CredentialPropertyDescriptors.USE_COMPUTE_ENGINE_CREDENTIALS, "true");
+ runner.assertValid();
+
+ Map<PropertyDescriptor, String> properties = runner.getProcessContext().getProperties();
+ final CredentialsFactory factory = new CredentialsFactory();
+ final GoogleCredentials credentials = factory.getGoogleCredentials(properties);
+
+ assertNotNull(credentials);
+ assertEquals("credentials class should be equal", ComputeEngineCredentials.class,
+ credentials.getClass());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/MockCredentialsFactoryProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/MockCredentialsFactoryProcessor.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/MockCredentialsFactoryProcessor.java
new file mode 100644
index 0000000..0734c40
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/factory/MockCredentialsFactoryProcessor.java
@@ -0,0 +1,74 @@
+/*
+ * 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.nifi.processors.gcp.credentials.factory;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.HttpServiceOptions;
+import com.google.cloud.Service;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.gcp.AbstractGCPProcessor;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.USE_APPLICATION_DEFAULT_CREDENTIALS;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.USE_COMPUTE_ENGINE_CREDENTIALS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class MockCredentialsFactoryProcessor extends AbstractGCPProcessor {
+ public final List<PropertyDescriptor> properties = Arrays.asList(
+ USE_APPLICATION_DEFAULT_CREDENTIALS,
+ SERVICE_ACCOUNT_JSON,
+ SERVICE_ACCOUNT_JSON_FILE,
+ USE_COMPUTE_ENGINE_CREDENTIALS
+ );
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ protected HttpServiceOptions getServiceOptions(ProcessContext context, GoogleCredentials credentials) {
+ HttpServiceOptions mockOptions = mock(HttpServiceOptions.class);
+ Service mockService = mock(Service.class);
+ when(mockOptions.getService()).thenReturn(mockService);
+
+ return mockOptions;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+
+ }
+
+ @Override
+ protected Collection<ValidationResult> customValidate(ValidationContext validationContext) {
+ CredentialsFactory factory = new CredentialsFactory();
+ return factory.validate(validationContext);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
new file mode 100644
index 0000000..edf0e0a
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/GCPCredentialsServiceTest.java
@@ -0,0 +1,154 @@
+/*
+ * 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.nifi.processors.gcp.credentials.service;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.auth.oauth2.ServiceAccountCredentials;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Test;
+
+import java.nio.file.Files;
+import java.nio.file.Paths;
+
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.SERVICE_ACCOUNT_JSON_FILE;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.USE_APPLICATION_DEFAULT_CREDENTIALS;
+import static org.apache.nifi.processors.gcp.credentials.factory.CredentialPropertyDescriptors.USE_COMPUTE_ENGINE_CREDENTIALS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+
+public class GCPCredentialsServiceTest {
+ @Test
+ public void testToString() throws Exception {
+ // toString method shouldn't cause an exception
+ final GCPCredentialsControllerService service = new GCPCredentialsControllerService();
+ service.toString();
+ }
+
+ @Test
+ public void testDefaultingToApplicationDefault() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.enableControllerService(serviceImpl);
+
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService) runner.getProcessContext()
+ .getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+ }
+
+ @Test
+ public void testExplicitApplicationDefault() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.setProperty(serviceImpl, USE_APPLICATION_DEFAULT_CREDENTIALS, "true");
+ runner.enableControllerService(serviceImpl);
+
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService) runner.getProcessContext()
+ .getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+ }
+
+ @Test
+ public void testFileCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.setProperty(serviceImpl, SERVICE_ACCOUNT_JSON_FILE,
+ "src/test/resources/mock-gcp-service-account.json");
+ runner.enableControllerService(serviceImpl);
+
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService) runner.getProcessContext()
+ .getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+
+ assertEquals("Credentials class should be equal", ServiceAccountCredentials.class,
+ credentials.getClass());
+ }
+
+ @Test
+ public void testBadFileCredentials() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+ runner.setProperty(serviceImpl, SERVICE_ACCOUNT_JSON_FILE,
+ "src/test/resources/bad-mock-gcp-service-account.json");
+ runner.assertNotValid(serviceImpl);
+ }
+
+ @Test
+ public void testMultipleCredentialSources() throws Exception {
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+ runner.setProperty(serviceImpl, SERVICE_ACCOUNT_JSON_FILE,
+ "src/test/resources/mock-gcp-service-account.json");
+ runner.setProperty(serviceImpl, USE_APPLICATION_DEFAULT_CREDENTIALS, "true");
+ runner.setProperty(serviceImpl, USE_COMPUTE_ENGINE_CREDENTIALS, "true");
+
+ runner.assertNotValid(serviceImpl);
+ }
+
+ @Test
+ public void testRawJsonCredentials() throws Exception {
+ final String jsonRead = new String(
+ Files.readAllBytes(Paths.get("src/test/resources/mock-gcp-service-account.json"))
+ );
+
+ final TestRunner runner = TestRunners.newTestRunner(MockCredentialsServiceProcessor.class);
+ final GCPCredentialsControllerService serviceImpl = new GCPCredentialsControllerService();
+ runner.addControllerService("gcpCredentialsProvider", serviceImpl);
+
+ runner.setProperty(serviceImpl, SERVICE_ACCOUNT_JSON,
+ jsonRead);
+ runner.enableControllerService(serviceImpl);
+
+ runner.assertValid(serviceImpl);
+
+ final GCPCredentialsService service = (GCPCredentialsService) runner.getProcessContext()
+ .getControllerServiceLookup().getControllerService("gcpCredentialsProvider");
+
+ assertNotNull(service);
+ final GoogleCredentials credentials = service.getGoogleCredentials();
+ assertNotNull(credentials);
+
+ assertEquals("Credentials class should be equal", ServiceAccountCredentials.class,
+ credentials.getClass());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/MockCredentialsServiceProcessor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/MockCredentialsServiceProcessor.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/MockCredentialsServiceProcessor.java
new file mode 100644
index 0000000..36c71d6
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/credentials/service/MockCredentialsServiceProcessor.java
@@ -0,0 +1,58 @@
+/*
+ * 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.nifi.processors.gcp.credentials.service;
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.HttpServiceOptions;
+import com.google.cloud.Service;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSession;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processors.gcp.AbstractGCPProcessor;
+
+import java.util.Arrays;
+import java.util.List;
+
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class MockCredentialsServiceProcessor extends AbstractGCPProcessor {
+ public final List<PropertyDescriptor> properties = Arrays.asList(
+ GCP_CREDENTIALS_PROVIDER_SERVICE
+ );
+
+ @Override
+ public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+ return properties;
+ }
+
+ @Override
+ protected HttpServiceOptions getServiceOptions(ProcessContext context, GoogleCredentials credentials) {
+ HttpServiceOptions mockOptions = mock(HttpServiceOptions.class);
+ Service mockService = mock(Service.class);
+ when(mockOptions.getService()).thenReturn(mockService);
+
+ return mockOptions;
+ }
+
+ @Override
+ public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java
new file mode 100644
index 0000000..725f824
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSIT.java
@@ -0,0 +1,177 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.BucketInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.gcp.GCPIntegrationTests;
+import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.experimental.categories.Category;
+
+import java.util.Arrays;
+import java.util.Iterator;
+
+import static org.junit.Assert.fail;
+
+/**
+ * Base class for GCS Integration Tests. Establishes a bucket and helper methods for creating test scenarios.
+ * Assumes use of <a href=https://developers.google.com/identity/protocols/application-default-credentials">Application Default</a>
+ * credentials for running tests.
+ */
+@Category(GCPIntegrationTests.class)
+public abstract class AbstractGCSIT {
+ private static final String PROJECT_ID = System.getProperty("test.gcp.project.id", "nifi-test-gcp-project");
+ protected static final String BUCKET = RemoteStorageHelper.generateBucketName();
+ protected static final String ENCRYPTION_KEY = "3gCN8OOPAGpDwRYieHAj6fR0eBSG5vloaHl9vlZ3doQ=";
+ protected static final Integer RETRIES = 6;
+
+ protected static RemoteStorageHelper helper;
+ protected static Storage storage;
+
+ @BeforeClass
+ public static void setUp() {
+ try {
+ helper = RemoteStorageHelper.create();
+ storage = helper.getOptions().getService();
+
+ if (storage.get(BUCKET) != null) {
+ // As the generateBucketName function uses a UUID, this should pretty much never happen
+ fail("Bucket " + BUCKET + " exists. Please rerun the test to generate a new bucket name.");
+ }
+
+ // Create the bucket
+ storage.create(BucketInfo.of(BUCKET));
+ } catch (StorageException e) {
+ fail("Can't create bucket " + BUCKET + ": " + e.getLocalizedMessage());
+ }
+
+ if (storage.get(BUCKET) == null) {
+ fail("Setup incomplete, tests will fail");
+ }
+ }
+
+ @AfterClass
+ public static void tearDown() {
+ try {
+ // Empty the bucket before deleting it.
+ Iterator<Blob> blobIterator = storage.list(BUCKET, Storage.BlobListOption.versions(true)).iterateAll();
+
+ while(blobIterator.hasNext()) {
+ Blob blob = blobIterator.next();
+ storage.delete(blob.getBlobId());
+ }
+
+ storage.delete(BUCKET);
+ } catch (final StorageException e) {
+ fail("Unable to delete bucket " + BUCKET + ": " + e.getLocalizedMessage());
+ }
+
+ if (storage.get(BUCKET) != null) {
+ fail("Incomplete teardown, subsequent tests might fail");
+ }
+ }
+
+ protected static TestRunner buildNewRunner(Processor processor) throws Exception {
+ final GCPCredentialsControllerService credentialsControllerService = new GCPCredentialsControllerService();
+ final TestRunner runner = TestRunners.newTestRunner(processor);
+ runner.addControllerService("gcpCredentialsControllerService", credentialsControllerService);
+ runner.enableControllerService(credentialsControllerService);
+
+ runner.setProperty(AbstractGCSProcessor.GCP_CREDENTIALS_PROVIDER_SERVICE, "gcpCredentialsControllerService");
+ runner.setProperty(AbstractGCSProcessor.PROJECT_ID, PROJECT_ID);
+ runner.setProperty(AbstractGCSProcessor.RETRY_COUNT, String.valueOf(RETRIES));
+
+ runner.assertValid(credentialsControllerService);
+
+ return runner;
+ }
+
+ /**
+ * Puts a test file onto Google Cloud Storage in bucket {@link AbstractGCSIT#BUCKET}.
+ *
+ * @param key Key which the file will be uploaded under
+ * @param bytes The content of the file to be uploaded
+ * @throws StorageException if the file can't be created for some reason
+ */
+ protected void putTestFile(String key, byte[] bytes) throws StorageException {
+ storage.create(BlobInfo.newBuilder(BlobId.of(BUCKET, key))
+ .build(), bytes
+ );
+ }
+
+ /**
+ * Puts a test file onto Google Cloud Storage in bucket {@link AbstractGCSIT#BUCKET}. This file is encrypted with
+ * server-side encryption using {@link AbstractGCSIT#ENCRYPTION_KEY}.
+ *
+ * @param key Key which the file will be uploaded under
+ * @param bytes The content of the file to be uploaded
+ * @throws StorageException if the file can't be created for some reason
+ */
+ protected void putTestFileEncrypted(String key, byte[] bytes) throws StorageException {
+ storage.create(BlobInfo.newBuilder(BlobId.of(BUCKET, key))
+ .build(), bytes, Storage.BlobTargetOption.encryptionKey(ENCRYPTION_KEY));
+ }
+
+ /**
+ * Test if the file exists in Google Cloud Storage in bucket {@link AbstractGCSIT#BUCKET}.
+ *
+ * @param key Key to check for the file
+ * @return true if the file exists, false if it doesn't
+ * @throws StorageException if there are any issues accessing the file or connecting to GCS.
+ */
+ protected boolean fileExists(String key) throws StorageException {
+ return (storage.get(BlobId.of(BUCKET, key)) != null);
+ }
+
+ /**
+ * Test if the file exists in Google Cloud Storage in bucket {@link AbstractGCSIT#BUCKET}, and if the content is as
+ * specified.
+ *
+ * @param key Key to check for the file
+ * @param bytes The content to compare to the content of the file
+ * @return true if the file exists and the content of the file is equal to {@code bytes}, false otherwise.
+ * @throws StorageException if there are any issues accessing the file or connecting to GCS.
+ */
+ protected boolean fileEquals(String key, byte[] bytes) throws StorageException {
+ return (fileExists(key) && Arrays.equals(storage.readAllBytes(BlobId.of(BUCKET, key)), bytes));
+ }
+
+ /**
+ * Test if the file exists in Google Cloud Storage in bucket {@link AbstractGCSIT#BUCKET}, and if the content is as
+ * specified. Assumes that the file is encrypted using {@link AbstractGCSIT#ENCRYPTION_KEY}.
+ *
+ * @param key Key to check for the file
+ * @param bytes The content to compare to the content of the file
+ * @return true if the file exists and the content of the file is equal to {@code bytes}, false otherwise.
+ * @throws StorageException if there are any issues accessing the file or connecting to GCS.
+ */
+ protected boolean fileEqualsEncrypted(String key, byte[] bytes) throws StorageException {
+ return (fileExists(key) && Arrays.equals(
+ storage.readAllBytes(BlobId.of(BUCKET, key), Storage.BlobSourceOption.decryptionKey(ENCRYPTION_KEY)),
+ bytes));
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSTest.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSTest.java
new file mode 100644
index 0000000..730cd85
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/AbstractGCSTest.java
@@ -0,0 +1,99 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+
+import com.google.auth.oauth2.GoogleCredentials;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageOptions;
+import com.google.cloud.storage.testing.RemoteStorageHelper;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsControllerService;
+import org.apache.nifi.processors.gcp.credentials.service.GCPCredentialsService;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+
+/**
+ * Base class for GCS Unit Tests. Provides a framework for creating a TestRunner instance with always-required credentials.
+ */
+public abstract class AbstractGCSTest {
+ private static final String PROJECT_ID = System.getProperty("test.gcp.project.id", "nifi-test-gcp-project");
+ private static final Integer RETRIES = 9;
+
+ static final String BUCKET = RemoteStorageHelper.generateBucketName();
+
+ @Before
+ public void setup() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ public static TestRunner buildNewRunner(Processor processor) throws Exception {
+ final GCPCredentialsService credentialsService = new GCPCredentialsControllerService();
+
+ final TestRunner runner = TestRunners.newTestRunner(processor);
+ runner.addControllerService("gcpCredentialsControllerService", credentialsService);
+ runner.enableControllerService(credentialsService);
+
+ runner.setProperty(AbstractGCSProcessor.GCP_CREDENTIALS_PROVIDER_SERVICE, "gcpCredentialsControllerService");
+ runner.setProperty(AbstractGCSProcessor.PROJECT_ID, PROJECT_ID);
+ runner.setProperty(AbstractGCSProcessor.RETRY_COUNT, String.valueOf(RETRIES));
+
+ runner.assertValid(credentialsService);
+
+ return runner;
+ }
+
+ public abstract AbstractGCSProcessor getProcessor();
+
+ protected abstract void addRequiredPropertiesToRunner(TestRunner runner);
+
+ @Mock
+ protected Storage storage;
+
+ @Test
+ public void testStorageOptionsConfiguration() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+
+ final AbstractGCSProcessor processor = getProcessor();
+ final GoogleCredentials mockCredentials = mock(GoogleCredentials.class);
+
+ final StorageOptions options = processor.getServiceOptions(runner.getProcessContext(),
+ mockCredentials);
+
+ assertEquals("Project IDs should match",
+ PROJECT_ID, options.getProjectId());
+
+ assertEquals("Retry counts should match",
+ RETRIES.intValue(), options.getRetryParams().getRetryMinAttempts());
+
+ assertEquals("Retry counts should match",
+ RETRIES.intValue(), options.getRetryParams().getRetryMaxAttempts());
+
+ assertSame("Credentials should be configured correctly",
+ mockCredentials, options.getCredentials());
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectIT.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectIT.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectIT.java
new file mode 100644
index 0000000..8beb521
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectIT.java
@@ -0,0 +1,91 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.util.TestRunner;
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Integration tests for {@link DeleteGCSObject} which actually use Google Cloud resources.
+ */
+public class DeleteGCSObjectIT extends AbstractGCSIT {
+ static final String KEY = "delete-me";
+
+ @Test
+ public void testSimpleDeleteWithFilename() throws Exception {
+ putTestFile(KEY, new byte[]{7, 8, 9});
+ assertTrue(fileExists(KEY));
+
+ final TestRunner runner = buildNewRunner(new DeleteGCSObject());
+ runner.setProperty(DeleteGCSObject.BUCKET, BUCKET);
+ runner.assertValid();
+
+ runner.enqueue("testdata", ImmutableMap.of(
+ CoreAttributes.FILENAME.key(), KEY
+ ));
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(DeleteGCSObject.REL_SUCCESS, 1);
+
+ assertFalse(fileExists(KEY));
+ }
+
+ @Test
+ public void testSimpleDeleteWithPropertySet() throws Exception {
+ putTestFile(KEY, new byte[]{7, 8, 9});
+ assertTrue(fileExists(KEY));
+
+ final TestRunner runner = buildNewRunner(new DeleteGCSObject());
+ runner.setProperty(DeleteGCSObject.BUCKET, BUCKET);
+ runner.setProperty(DeleteGCSObject.KEY, KEY);
+ runner.assertValid();
+
+ runner.enqueue("testdata", ImmutableMap.of(
+ "filename", "different-filename"
+ ));
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(DeleteGCSObject.REL_SUCCESS, 1);
+
+ assertFalse(fileExists(KEY));
+ }
+
+ @Test
+ public void testDeleteNonExistentFile() throws Exception {
+ final TestRunner runner = buildNewRunner(new DeleteGCSObject());
+ runner.setProperty(DeleteGCSObject.BUCKET, BUCKET);
+ runner.assertValid();
+
+ runner.enqueue("testdata", ImmutableMap.of(
+ "filename", "nonexistant-file"
+ ));
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(DeleteGCSObject.REL_SUCCESS, 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectTest.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectTest.java
new file mode 100644
index 0000000..bbf97b8
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/DeleteGCSObjectTest.java
@@ -0,0 +1,148 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.common.collect.ImmutableMap;
+import org.apache.nifi.util.TestRunner;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link DeleteGCSObject}. No connections to the Google Cloud service are made.
+ */
+public class DeleteGCSObjectTest extends AbstractGCSTest {
+ public static final Long GENERATION = 42L;
+ static final String KEY = "somefile";
+
+
+ public static final String BUCKET_ATTR = "gcs.bucket";
+ public static final String KEY_ATTR = "gcs.key";
+ public static final String GENERATION_ATTR = "gcs.generation";
+
+ @Mock
+ Storage storage;
+
+ @Before
+ public void setup() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Override
+ protected void addRequiredPropertiesToRunner(TestRunner runner) {
+ runner.setProperty(DeleteGCSObject.BUCKET, BUCKET);
+ runner.setProperty(DeleteGCSObject.GENERATION, String.valueOf(GENERATION));
+ runner.setProperty(DeleteGCSObject.KEY, KEY);
+ }
+
+ @Override
+ public DeleteGCSObject getProcessor() {
+ return new DeleteGCSObject() {
+ @Override
+ protected Storage getCloudService() {
+ return storage;
+ }
+ };
+ }
+
+
+ @Test
+ public void testDeleteWithValidArguments() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+
+ runner.assertValid();
+
+ runner.enqueue("testdata");
+
+ runner.run();
+
+ verify(storage).delete(eq(BlobId.of(BUCKET, KEY, GENERATION)));
+
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(DeleteGCSObject.REL_SUCCESS, 1);
+ }
+
+ @Test
+ public void testTwoDeletesWithFlowfileAttributes() throws Exception {
+ reset(storage);
+
+ final TestRunner runner = buildNewRunner(getProcessor());
+
+ runner.setProperty(DeleteGCSObject.BUCKET, "${" + BUCKET_ATTR + "}");
+ runner.setProperty(DeleteGCSObject.KEY, "${" + KEY_ATTR + "}");
+ runner.setProperty(DeleteGCSObject.GENERATION, "${" + GENERATION_ATTR + "}");
+
+ runner.assertValid();
+
+ final String bucket1 = BUCKET + "_1";
+ final String bucket2 = BUCKET + "_2";
+ final String key1 = KEY + "_1";
+ final String key2 = KEY + "_2";
+ final Long generation1 = GENERATION + 1L;
+ final Long generation2 = GENERATION + 2L;
+
+ runner.enqueue("testdata1", ImmutableMap.of(
+ BUCKET_ATTR, bucket1,
+ KEY_ATTR, key1,
+ GENERATION_ATTR, String.valueOf(generation1)
+ ));
+
+ runner.enqueue("testdata2", ImmutableMap.of(
+ BUCKET_ATTR, bucket2,
+ KEY_ATTR, key2,
+ GENERATION_ATTR, String.valueOf(generation2)
+ ));
+
+ runner.run(2);
+
+ verify(storage).delete(eq(BlobId.of(bucket1, key1, generation1)));
+ verify(storage).delete(eq(BlobId.of(bucket2, key2, generation2)));
+
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(DeleteGCSObject.REL_SUCCESS, 2);
+ }
+
+ @Test
+ public void testFailureOnException() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ runner.enqueue("testdata");
+
+ when(storage.delete(any(BlobId.class))).thenThrow(new StorageException(1, "Test Exception"));
+
+ runner.run();
+
+ runner.assertPenalizeCount(1);
+ runner.assertAllFlowFilesTransferred(DeleteGCSObject.REL_FAILURE);
+ runner.assertTransferCount(DeleteGCSObject.REL_FAILURE, 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java
new file mode 100644
index 0000000..a8b8306
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectIT.java
@@ -0,0 +1,102 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.common.collect.ImmutableMap;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Integration tests for {@link FetchGCSObject} which actually use Google Cloud resources.
+ */
+public class FetchGCSObjectIT extends AbstractGCSIT {
+ static final String KEY = "delete-me";
+ static final byte[] CONTENT = {10,11,12};
+
+ @Test
+ public void testSimpleFetch() throws Exception {
+ putTestFile(KEY, CONTENT);
+ assertTrue(fileExists(KEY));
+
+ final TestRunner runner = buildNewRunner(new FetchGCSObject());
+ runner.setProperty(FetchGCSObject.BUCKET, BUCKET);
+
+ runner.enqueue(new byte[0], ImmutableMap.of(
+ "filename", KEY
+ ));
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS, 1);
+ final List<MockFlowFile> ffs = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS);
+ MockFlowFile ff = ffs.get(0);
+
+ ff.assertContentEquals(CONTENT);
+
+ ff.assertAttributeNotExists(StorageAttributes.ENCRYPTION_ALGORITHM_ATTR);
+ ff.assertAttributeNotExists(StorageAttributes.ENCRYPTION_SHA256_ATTR);
+
+ for (final Map.Entry<String, String> entry : ff.getAttributes().entrySet()) {
+ System.out.println(entry.getKey() + ":" + entry.getValue());
+ }
+ }
+
+ @Test
+ public void testSimpleFetchEncrypted() throws Exception {
+ putTestFileEncrypted(KEY, CONTENT);
+ assertTrue(fileExists(KEY));
+
+ final TestRunner runner = buildNewRunner(new FetchGCSObject());
+ runner.setProperty(FetchGCSObject.BUCKET, BUCKET);
+ runner.setProperty(FetchGCSObject.ENCRYPTION_KEY, ENCRYPTION_KEY);
+
+ runner.enqueue(new byte[0], ImmutableMap.of(
+ "filename", KEY
+ ));
+
+ runner.assertValid();
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS, 1);
+ final List<MockFlowFile> ffs = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS);
+ MockFlowFile ff = ffs.get(0);
+
+ ff.assertAttributeEquals(StorageAttributes.ENCRYPTION_ALGORITHM_ATTR,
+ "AES256");
+ }
+
+
+ @Test
+ public void testFetchNonexistantFile() throws Exception {
+ final TestRunner runner = buildNewRunner(new FetchGCSObject());
+ runner.setProperty(FetchGCSObject.BUCKET, BUCKET);
+ runner.enqueue(new byte[0], ImmutableMap.of(
+ "filename", "non-existent"
+ ));
+
+ runner.assertValid();
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_FAILURE, 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectTest.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectTest.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectTest.java
new file mode 100644
index 0000000..0d8b209
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/FetchGCSObjectTest.java
@@ -0,0 +1,614 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.cloud.ReadChannel;
+import com.google.cloud.RestorableState;
+import com.google.cloud.storage.Acl;
+import com.google.cloud.storage.Blob;
+import com.google.cloud.storage.BlobId;
+import com.google.cloud.storage.BlobInfo;
+import com.google.cloud.storage.Storage;
+import com.google.cloud.storage.StorageException;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Set;
+
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.BUCKET_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CACHE_CONTROL_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.COMPONENT_COUNT_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CONTENT_DISPOSITION_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CONTENT_ENCODING_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CONTENT_LANGUAGE_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CRC32C_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.CREATE_TIME_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.ENCRYPTION_ALGORITHM_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.ENCRYPTION_SHA256_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.ETAG_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.GENERATED_ID_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.GENERATION_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.KEY_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.MD5_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.MEDIA_LINK_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.METAGENERATION_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.OWNER_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.OWNER_TYPE_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.UPDATE_TIME_ATTR;
+import static org.apache.nifi.processors.gcp.storage.StorageAttributes.URI_ATTR;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link FetchGCSObject}.
+ */
+public class FetchGCSObjectTest extends AbstractGCSTest {
+ private final static String KEY = "test-key";
+ private final static Long GENERATION = 5L;
+ private static final String CONTENT = "test-content";
+
+ private static final Long SIZE = 100L;
+ private static final String CACHE_CONTROL = "test-cache-control";
+ private static final Integer COMPONENT_COUNT = 3;
+ private static final String CONTENT_ENCODING = "test-content-encoding";
+ private static final String CONTENT_LANGUAGE = "test-content-language";
+ private static final String CONTENT_TYPE = "test-content-type";
+ private static final String CRC32C = "test-crc32c";
+ private static final String ENCRYPTION = "test-encryption";
+ private static final String ENCRYPTION_SHA256 = "test-encryption-256";
+ private static final String ETAG = "test-etag";
+ private static final String GENERATED_ID = "test-generated-id";
+ private static final String MD5 = "test-md5";
+ private static final String MEDIA_LINK = "test-media-link";
+ private static final Long METAGENERATION = 42L;
+ private static final String OWNER_USER_EMAIL = "test-owner-user-email";
+ private static final String OWNER_GROUP_EMAIL = "test-owner-group-email";
+ private static final String OWNER_DOMAIN = "test-owner-domain";
+ private static final String OWNER_PROJECT_ID = "test-owner-project-id";
+ private static final String URI = "test-uri";
+ private static final String CONTENT_DISPOSITION = "attachment; filename=\"test-content-disposition.txt\"";
+ private static final Long CREATE_TIME = 1234L;
+ private static final Long UPDATE_TIME = 4567L;
+
+
+ @Mock
+ Storage storage;
+
+ @Before
+ public void setup() throws Exception {
+ MockitoAnnotations.initMocks(this);
+ }
+
+ @Override
+ public AbstractGCSProcessor getProcessor() {
+ return new FetchGCSObject() {
+ @Override
+ protected Storage getCloudService() {
+ return storage;
+ }
+ };
+ }
+
+ private class MockReadChannel implements ReadChannel {
+ private byte[] toRead;
+ private int position = 0;
+ private boolean finished;
+ private boolean isOpen;
+
+ private MockReadChannel(String textToRead) {
+ this.toRead = textToRead.getBytes();
+ this.isOpen = true;
+ this.finished = false;
+ }
+
+ @Override
+ public void close() {
+ this.isOpen = false;
+ }
+
+ @Override
+ public void seek(long l) throws IOException {
+
+ }
+
+ @Override
+ public void chunkSize(int i) {
+
+ }
+
+ @Override
+ public void setChunkSize(int i) {
+
+ }
+
+ @Override
+ public RestorableState<ReadChannel> capture() {
+ return null;
+ }
+
+ @Override
+ public int read(ByteBuffer dst) throws IOException {
+ if (this.finished) {
+ return -1;
+ } else {
+ if (dst.remaining() > this.toRead.length) {
+ this.finished = true;
+ }
+ int toWrite = Math.min(this.toRead.length - position, dst.remaining());
+
+ dst.put(this.toRead, this.position, toWrite);
+ this.position += toWrite;
+
+ return toWrite;
+ }
+ }
+
+ @Override
+ public boolean isOpen() {
+ return this.isOpen;
+ }
+ }
+
+ @Override
+ protected void addRequiredPropertiesToRunner(TestRunner runner) {
+ runner.setProperty(FetchGCSObject.BUCKET, BUCKET);
+ runner.setProperty(FetchGCSObject.KEY, String.valueOf(KEY));
+ }
+
+ @Test
+ public void testSuccessfulFetch() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+
+ when(blob.getBucket()).thenReturn(BUCKET);
+ when(blob.getName()).thenReturn(KEY);
+ when(blob.getSize()).thenReturn(SIZE);
+ when(blob.getCacheControl()).thenReturn(CACHE_CONTROL);
+ when(blob.getComponentCount()).thenReturn(COMPONENT_COUNT);
+ when(blob.getContentEncoding()).thenReturn(CONTENT_ENCODING);
+ when(blob.getContentLanguage()).thenReturn(CONTENT_LANGUAGE);
+ when(blob.getContentType()).thenReturn(CONTENT_TYPE);
+ when(blob.getCrc32c()).thenReturn(CRC32C);
+
+ final BlobInfo.CustomerEncryption mockEncryption = mock(BlobInfo.CustomerEncryption.class);
+ when(mockEncryption.getEncryptionAlgorithm()).thenReturn(ENCRYPTION);
+ when(mockEncryption.getKeySha256()).thenReturn(ENCRYPTION_SHA256);
+ when(blob.getCustomerEncryption()).thenReturn(mockEncryption);
+
+ when(blob.getEtag()).thenReturn(ETAG);
+ when(blob.getGeneratedId()).thenReturn(GENERATED_ID);
+ when(blob.getGeneration()).thenReturn(GENERATION);
+ when(blob.getMd5()).thenReturn(MD5);
+ when(blob.getMediaLink()).thenReturn(MEDIA_LINK);
+ when(blob.getMetageneration()).thenReturn(METAGENERATION);
+ when(blob.getSelfLink()).thenReturn(URI);
+ when(blob.getContentDisposition()).thenReturn(CONTENT_DISPOSITION);
+ when(blob.getCreateTime()).thenReturn(CREATE_TIME);
+ when(blob.getUpdateTime()).thenReturn(UPDATE_TIME);
+
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+
+ runner.enqueue("");
+
+ runner.run();
+
+ verify(storage).get(any(BlobId.class));
+ verify(storage).reader(any(BlobId.class), any(Storage.BlobSourceOption.class));
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(FetchGCSObject.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS).get(0);
+
+ assertTrue(flowFile.isContentEqual(CONTENT));
+ assertEquals(
+ BUCKET,
+ flowFile.getAttribute(BUCKET_ATTR)
+ );
+
+ assertEquals(
+ KEY,
+ flowFile.getAttribute(KEY_ATTR)
+ );
+
+ assertEquals(
+ CACHE_CONTROL,
+ flowFile.getAttribute(CACHE_CONTROL_ATTR)
+ );
+
+ assertEquals(
+ COMPONENT_COUNT,
+ Integer.valueOf(flowFile.getAttribute(COMPONENT_COUNT_ATTR))
+ );
+
+ assertEquals(
+ CONTENT_ENCODING,
+ flowFile.getAttribute(CONTENT_ENCODING_ATTR)
+ );
+
+ assertEquals(
+ CONTENT_LANGUAGE,
+ flowFile.getAttribute(CONTENT_LANGUAGE_ATTR)
+ );
+
+ assertEquals(
+ CONTENT_TYPE,
+ flowFile.getAttribute(CoreAttributes.MIME_TYPE.key())
+ );
+
+ assertEquals(
+ CRC32C,
+ flowFile.getAttribute(CRC32C_ATTR)
+ );
+
+ assertEquals(
+ ENCRYPTION,
+ flowFile.getAttribute(ENCRYPTION_ALGORITHM_ATTR)
+ );
+
+ assertEquals(
+ ENCRYPTION_SHA256,
+ flowFile.getAttribute(ENCRYPTION_SHA256_ATTR)
+ );
+
+ assertEquals(
+ ETAG,
+ flowFile.getAttribute(ETAG_ATTR)
+ );
+
+ assertEquals(
+ GENERATED_ID,
+ flowFile.getAttribute(GENERATED_ID_ATTR)
+ );
+
+ assertEquals(
+ GENERATION,
+ Long.valueOf(flowFile.getAttribute(GENERATION_ATTR))
+ );
+
+ assertEquals(
+ MD5,
+ flowFile.getAttribute(MD5_ATTR)
+ );
+
+ assertEquals(
+ MEDIA_LINK,
+ flowFile.getAttribute(MEDIA_LINK_ATTR)
+ );
+
+ assertEquals(
+ METAGENERATION,
+ Long.valueOf(flowFile.getAttribute(METAGENERATION_ATTR))
+ );
+
+ assertEquals(
+ URI,
+ flowFile.getAttribute(URI_ATTR)
+ );
+
+ assertEquals(
+ CONTENT_DISPOSITION,
+ flowFile.getAttribute(CONTENT_DISPOSITION_ATTR)
+ );
+
+ assertEquals(
+ CREATE_TIME,
+ Long.valueOf(flowFile.getAttribute(CREATE_TIME_ATTR))
+ );
+
+ assertEquals(
+ UPDATE_TIME,
+ Long.valueOf(flowFile.getAttribute(UPDATE_TIME_ATTR))
+ );
+
+ }
+
+ @Test
+ public void testAclOwnerUser() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+
+ final Acl.User mockUser = mock(Acl.User.class);
+ when(mockUser.getEmail()).thenReturn(OWNER_USER_EMAIL);
+ when(blob.getOwner()).thenReturn(mockUser);
+
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+ runner.enqueue("");
+
+ runner.run();
+
+ verify(storage).get(any(BlobId.class));
+ verify(storage).reader(any(BlobId.class), any(Storage.BlobSourceOption.class));
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(FetchGCSObject.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS).get(0);
+
+ assertEquals(
+ OWNER_USER_EMAIL,
+ flowFile.getAttribute(OWNER_ATTR)
+ );
+
+ assertEquals(
+ "user",
+ flowFile.getAttribute(OWNER_TYPE_ATTR)
+ );
+ }
+
+ @Test
+ public void testAclOwnerGroup() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+
+ final Acl.Group mockGroup = mock(Acl.Group.class);
+ when(mockGroup.getEmail()).thenReturn(OWNER_GROUP_EMAIL);
+ when(blob.getOwner()).thenReturn(mockGroup);
+
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+
+ runner.enqueue("");
+
+ runner.run();
+
+ verify(storage).get(any(BlobId.class));
+ verify(storage).reader(any(BlobId.class), any(Storage.BlobSourceOption.class));
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(FetchGCSObject.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS).get(0);
+
+ assertEquals(
+ OWNER_GROUP_EMAIL,
+ flowFile.getAttribute(OWNER_ATTR)
+ );
+
+ assertEquals(
+ "group",
+ flowFile.getAttribute(OWNER_TYPE_ATTR)
+ );
+ }
+
+
+ @Test
+ public void testAclOwnerDomain() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+
+ final Acl.Domain mockDomain = mock(Acl.Domain.class);
+ when(mockDomain.getDomain()).thenReturn(OWNER_DOMAIN);
+ when(blob.getOwner()).thenReturn(mockDomain);
+
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+
+ runner.enqueue("");
+
+ runner.run();
+
+ verify(storage).get(any(BlobId.class));
+ verify(storage).reader(any(BlobId.class), any(Storage.BlobSourceOption.class));
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(FetchGCSObject.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS).get(0);
+
+ assertEquals(
+ OWNER_DOMAIN,
+ flowFile.getAttribute(OWNER_ATTR)
+ );
+
+ assertEquals(
+ "domain",
+ flowFile.getAttribute(OWNER_TYPE_ATTR)
+ );
+ }
+
+
+ @Test
+ public void testAclOwnerProject() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+ final Acl.Project mockProject = mock(Acl.Project.class);
+ when(mockProject.getProjectId()).thenReturn(OWNER_PROJECT_ID);
+ when(blob.getOwner()).thenReturn(mockProject);
+
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+ runner.enqueue("");
+
+ runner.run();
+
+ verify(storage).get(any(BlobId.class));
+ verify(storage).reader(any(BlobId.class), any(Storage.BlobSourceOption.class));
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_SUCCESS);
+ runner.assertTransferCount(FetchGCSObject.REL_SUCCESS, 1);
+ final MockFlowFile flowFile = runner.getFlowFilesForRelationship(FetchGCSObject.REL_SUCCESS).get(0);
+
+ assertEquals(
+ OWNER_PROJECT_ID,
+ flowFile.getAttribute(OWNER_ATTR)
+ );
+
+ assertEquals(
+ "project",
+ flowFile.getAttribute(OWNER_TYPE_ATTR)
+ );
+ }
+
+ @Test
+ public void testBlobIdWithGeneration() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+
+ runner.removeProperty(FetchGCSObject.KEY);
+ runner.removeProperty(FetchGCSObject.BUCKET);
+
+ runner.setProperty(FetchGCSObject.GENERATION, String.valueOf(GENERATION));
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+ runner.enqueue("", ImmutableMap.of(
+ BUCKET_ATTR, BUCKET,
+ CoreAttributes.FILENAME.key(), KEY
+ ));
+
+ runner.run();
+
+ ArgumentCaptor<BlobId> blobIdArgumentCaptor = ArgumentCaptor.forClass(BlobId.class);
+ ArgumentCaptor<Storage.BlobSourceOption> blobSourceOptionArgumentCaptor = ArgumentCaptor.forClass(Storage.BlobSourceOption.class);
+ verify(storage).get(blobIdArgumentCaptor.capture());
+ verify(storage).reader(any(BlobId.class), blobSourceOptionArgumentCaptor.capture());
+
+ final BlobId blobId = blobIdArgumentCaptor.getValue();
+
+ assertEquals(
+ BUCKET,
+ blobId.getBucket()
+ );
+
+ assertEquals(
+ KEY,
+ blobId.getName()
+ );
+
+ assertEquals(
+ GENERATION,
+ blobId.getGeneration()
+ );
+
+
+ final Set<Storage.BlobSourceOption> blobSourceOptions = ImmutableSet.copyOf(blobSourceOptionArgumentCaptor.getAllValues());
+ assertTrue(blobSourceOptions.contains(Storage.BlobSourceOption.generationMatch()));
+ assertEquals(
+ 1,
+ blobSourceOptions.size()
+ );
+
+ }
+
+
+ @Test
+ public void testBlobIdWithEncryption() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+
+ runner.setProperty(FetchGCSObject.ENCRYPTION_KEY, ENCRYPTION_SHA256);
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ final Blob blob = mock(Blob.class);
+ when(storage.get(any(BlobId.class))).thenReturn(blob);
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+ runner.enqueue("");
+
+ runner.run();
+
+ ArgumentCaptor<BlobId> blobIdArgumentCaptor = ArgumentCaptor.forClass(BlobId.class);
+ ArgumentCaptor<Storage.BlobSourceOption> blobSourceOptionArgumentCaptor = ArgumentCaptor.forClass(Storage.BlobSourceOption.class);
+ verify(storage).get(blobIdArgumentCaptor.capture());
+ verify(storage).reader(any(BlobId.class), blobSourceOptionArgumentCaptor.capture());
+
+ final BlobId blobId = blobIdArgumentCaptor.getValue();
+
+ assertEquals(
+ BUCKET,
+ blobId.getBucket()
+ );
+
+ assertEquals(
+ KEY,
+ blobId.getName()
+ );
+
+ assertNull(blobId.getGeneration());
+
+ final Set<Storage.BlobSourceOption> blobSourceOptions = ImmutableSet.copyOf(blobSourceOptionArgumentCaptor.getAllValues());
+
+ assertTrue(blobSourceOptions.contains(Storage.BlobSourceOption.decryptionKey(ENCRYPTION_SHA256)));
+ assertEquals(
+ 1,
+ blobSourceOptions.size()
+ );
+ }
+
+ @Test
+ public void testStorageExceptionOnFetch() throws Exception {
+ reset(storage);
+ final TestRunner runner = buildNewRunner(getProcessor());
+ addRequiredPropertiesToRunner(runner);
+ runner.assertValid();
+
+ when(storage.get(any(BlobId.class))).thenThrow(new StorageException(400, "test-exception"));
+ when(storage.reader(any(BlobId.class), any(Storage.BlobSourceOption.class))).thenReturn(new MockReadChannel(CONTENT));
+
+ runner.enqueue("");
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(FetchGCSObject.REL_FAILURE);
+ runner.assertTransferCount(FetchGCSObject.REL_FAILURE, 1);
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/897c7029/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/ListGCSBucketIT.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/ListGCSBucketIT.java b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/ListGCSBucketIT.java
new file mode 100644
index 0000000..497e8e0
--- /dev/null
+++ b/nifi-nar-bundles/nifi-gcp-bundle/nifi-gcp-processors/src/test/java/org/apache/nifi/processors/gcp/storage/ListGCSBucketIT.java
@@ -0,0 +1,132 @@
+/*
+ * 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.nifi.processors.gcp.storage;
+
+import com.google.cloud.storage.BucketInfo;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.TestRunner;
+import org.junit.Test;
+
+import java.util.List;
+
+import static org.junit.Assert.assertNotEquals;
+
+/**
+ * Integration tests for {@link ListGCSBucket} which actually use Google Cloud resources.
+ */
+public class ListGCSBucketIT extends AbstractGCSIT {
+ private static final byte[] CONTENT = {12, 13, 14};
+
+ @Test
+ public void testSimpleList() throws Exception {
+ putTestFile("a", CONTENT);
+ putTestFile("b/c", CONTENT);
+ putTestFile("d/e", CONTENT);
+
+ final TestRunner runner = buildNewRunner(new ListGCSBucket());
+ runner.setProperty(ListGCSBucket.BUCKET, BUCKET);
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ListGCSBucket.REL_SUCCESS, 3);
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListGCSBucket.REL_SUCCESS);
+ flowFiles.get(0).assertAttributeEquals("filename", "a");
+ flowFiles.get(1).assertAttributeEquals("filename", "b/c");
+ flowFiles.get(2).assertAttributeEquals("filename", "d/e");
+ }
+
+
+ @Test
+ public void testSimpleListWithPrefix() throws Exception {
+ putTestFile("a", CONTENT);
+ putTestFile("b/c", CONTENT);
+ putTestFile("d/e", CONTENT);
+
+ final TestRunner runner = buildNewRunner(new ListGCSBucket());
+ runner.setProperty(ListGCSBucket.BUCKET, BUCKET);
+
+ runner.setProperty(ListGCSBucket.PREFIX, "b/");
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ListGCSBucket.REL_SUCCESS, 1);
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListGCSBucket.REL_SUCCESS);
+ flowFiles.get(0).assertAttributeEquals("filename", "b/c");
+ }
+
+
+
+ @Test
+ public void testSimpleListWithPrefixAndGenerations() throws Exception {
+ // enable versioning
+ storage.update(BucketInfo.newBuilder(BUCKET).setVersioningEnabled(true).build());
+
+ putTestFile("generations/a", CONTENT);
+ putTestFile("generations/a", CONTENT);
+ putTestFile("generations/b", CONTENT);
+ putTestFile("generations/c", CONTENT);
+
+ final TestRunner runner = buildNewRunner(new ListGCSBucket());
+ runner.setProperty(ListGCSBucket.BUCKET, BUCKET);
+
+ runner.setProperty(ListGCSBucket.PREFIX, "generations/");
+ runner.setProperty(ListGCSBucket.USE_GENERATIONS, "true");
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ListGCSBucket.REL_SUCCESS, 4);
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListGCSBucket.REL_SUCCESS);
+ flowFiles.get(0).assertAttributeEquals("filename", "generations/a");
+ flowFiles.get(1).assertAttributeEquals("filename", "generations/a");
+ flowFiles.get(2).assertAttributeEquals("filename", "generations/b");
+ flowFiles.get(3).assertAttributeEquals("filename", "generations/c");
+
+ assertNotEquals(
+ flowFiles.get(0).getAttribute(StorageAttributes.GENERATION_ATTR),
+ flowFiles.get(1).getAttribute(StorageAttributes.GENERATION_ATTR)
+ );
+ }
+
+
+ @Test
+ public void testCheckpointing() throws Exception {
+ putTestFile("checkpoint/a", CONTENT);
+ putTestFile("checkpoint/b/c", CONTENT);
+
+ final TestRunner runner = buildNewRunner(new ListGCSBucket());
+ runner.setProperty(ListGCSBucket.BUCKET, BUCKET);
+
+ runner.setProperty(ListGCSBucket.PREFIX, "checkpoint/");
+
+ runner.run();
+
+ runner.assertAllFlowFilesTransferred(ListGCSBucket.REL_SUCCESS, 2);
+ List<MockFlowFile> flowFiles = runner.getFlowFilesForRelationship(ListGCSBucket.REL_SUCCESS);
+ flowFiles.get(0).assertAttributeEquals("filename", "checkpoint/a");
+ flowFiles.get(1).assertAttributeEquals("filename", "checkpoint/b/c");
+
+ putTestFile("checkpoint/d/e", CONTENT);
+ runner.run();
+
+ // Should only retrieve 1 new file (for a total of 3)
+ runner.assertAllFlowFilesTransferred(ListGCSBucket.REL_SUCCESS, 3);
+ flowFiles = runner.getFlowFilesForRelationship(ListGCSBucket.REL_SUCCESS);
+ flowFiles.get(0).assertAttributeEquals("filename", "checkpoint/a");
+ flowFiles.get(1).assertAttributeEquals("filename", "checkpoint/b/c");
+ flowFiles.get(2).assertAttributeEquals("filename", "checkpoint/d/e");
+ }
+}