mirror of https://github.com/apache/nifi.git
NIFI-11117 Remove folder creation from PutGoogleDrive
This closes #6910. Signed-off-by: Peter Turcsanyi <turcsanyi@apache.org>
This commit is contained in:
krisztina-zsihovszki
Peter Turcsanyi
parent
5daf01e760
commit
92ccb79670
|
|
@ -65,6 +65,11 @@
|
|||
<version>1.20.0-SNAPSHOT</version>
|
||||
<scope>provided</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.nifi</groupId>
|
||||
<artifactId>nifi-conflict-resolution</artifactId>
|
||||
<version>1.20.0-SNAPSHOT</version>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.apache.nifi</groupId>
|
||||
<artifactId>nifi-mock</artifactId>
|
||||
|
|
|
|||
|
|
@ -183,7 +183,7 @@ public class FetchGoogleDrive extends AbstractProcessor implements GoogleDriveTr
|
|||
}
|
||||
|
||||
private void handleErrorResponse(ProcessSession session, String fileId, FlowFile flowFile, GoogleJsonResponseException e) {
|
||||
getLogger().error("Couldn't fetch file with id '{}'", fileId, e);
|
||||
getLogger().error("Fetching File [{}] failed", fileId, e);
|
||||
|
||||
flowFile = session.putAttribute(flowFile, ERROR_CODE, "" + e.getStatusCode());
|
||||
flowFile = session.putAttribute(flowFile, ERROR_MESSAGE, e.getMessage());
|
||||
|
|
@ -193,7 +193,7 @@ public class FetchGoogleDrive extends AbstractProcessor implements GoogleDriveTr
|
|||
}
|
||||
|
||||
private void handleUnexpectedError(ProcessSession session, FlowFile flowFile, String fileId, Exception e) {
|
||||
getLogger().error("Unexpected error while fetching and processing file with id '{}'", fileId, e);
|
||||
getLogger().error("Fetching File [{}] failed", fileId, e);
|
||||
|
||||
flowFile = session.putAttribute(flowFile, ERROR_MESSAGE, e.getMessage());
|
||||
|
||||
|
|
|
|||
|
|
@ -23,7 +23,8 @@ import static java.util.Arrays.asList;
|
|||
import static java.util.Collections.singletonList;
|
||||
import static java.util.stream.Collectors.joining;
|
||||
import static org.apache.nifi.processor.util.StandardValidators.DATA_SIZE_VALIDATOR;
|
||||
import static org.apache.nifi.processor.util.StandardValidators.createRegexMatchingValidator;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.FAIL;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
|
||||
import static org.apache.nifi.processors.gcp.drive.GoogleDriveAttributes.ERROR_CODE;
|
||||
import static org.apache.nifi.processors.gcp.drive.GoogleDriveAttributes.ERROR_CODE_DESC;
|
||||
import static org.apache.nifi.processors.gcp.drive.GoogleDriveAttributes.ERROR_MESSAGE;
|
||||
|
|
@ -66,7 +67,6 @@ import java.util.Map;
|
|||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.regex.Pattern;
|
||||
import org.apache.nifi.annotation.behavior.InputRequirement;
|
||||
import org.apache.nifi.annotation.behavior.InputRequirement.Requirement;
|
||||
import org.apache.nifi.annotation.behavior.ReadsAttribute;
|
||||
|
|
@ -90,6 +90,7 @@ import org.apache.nifi.processor.ProcessSession;
|
|||
import org.apache.nifi.processor.Relationship;
|
||||
import org.apache.nifi.processor.exception.ProcessException;
|
||||
import org.apache.nifi.processor.util.StandardValidators;
|
||||
import org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy;
|
||||
import org.apache.nifi.processors.gcp.ProxyAwareTransportFactory;
|
||||
import org.apache.nifi.proxy.ProxyConfiguration;
|
||||
import org.json.JSONObject;
|
||||
|
|
@ -108,10 +109,6 @@ import org.json.JSONObject;
|
|||
@WritesAttribute(attribute = ERROR_CODE, description = ERROR_CODE_DESC),
|
||||
@WritesAttribute(attribute = ERROR_MESSAGE, description = ERROR_MESSAGE_DESC)})
|
||||
public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrait {
|
||||
|
||||
public static final String IGNORE_RESOLUTION = "ignore";
|
||||
public static final String REPLACE_RESOLUTION = "replace";
|
||||
public static final String FAIL_RESOLUTION = "fail";
|
||||
public static final int MIN_ALLOWED_CHUNK_SIZE_IN_BYTES = MediaHttpUploader.MINIMUM_CHUNK_SIZE;
|
||||
public static final int MAX_ALLOWED_CHUNK_SIZE_IN_BYTES = 1024 * 1024 * 1024;
|
||||
|
||||
|
|
@ -125,17 +122,6 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
.required(true)
|
||||
.build();
|
||||
|
||||
public static final PropertyDescriptor SUBFOLDER_NAME = new PropertyDescriptor.Builder()
|
||||
.name("subfolder-name")
|
||||
.displayName("Subfolder Name")
|
||||
.description("The name (path) of the subfolder where files are uploaded. The subfolder name is relative to the shared folder specified by 'Folder ID'."
|
||||
+ " Example: subFolder, subFolder1/subfolder2")
|
||||
.addValidator(createRegexMatchingValidator(Pattern.compile("^(?!/).+(?<!/)$"), false,
|
||||
"Subfolder Name should not contain leading or trailing slash ('/') character."))
|
||||
.expressionLanguageSupported(ExpressionLanguageScope.FLOWFILE_ATTRIBUTES)
|
||||
.required(false)
|
||||
.build();
|
||||
|
||||
public static final PropertyDescriptor FILE_NAME = new PropertyDescriptor.Builder()
|
||||
.name("file-name")
|
||||
.displayName("Filename")
|
||||
|
|
@ -146,26 +132,13 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
.addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
|
||||
.build();
|
||||
|
||||
public static final PropertyDescriptor CREATE_SUBFOLDER = new PropertyDescriptor.Builder()
|
||||
.name("create-subfolder")
|
||||
.displayName("Create Subfolder")
|
||||
.expressionLanguageSupported(ExpressionLanguageScope.NONE)
|
||||
.required(true)
|
||||
.addValidator(StandardValidators.BOOLEAN_VALIDATOR)
|
||||
.allowableValues("true", "false")
|
||||
.defaultValue("false")
|
||||
.dependsOn(SUBFOLDER_NAME)
|
||||
.description("Specifies whether to automatically create the subfolder specified by 'Folder Name' if it does not exist. " +
|
||||
"Permission to list folders is required. ")
|
||||
.build();
|
||||
|
||||
public static final PropertyDescriptor CONFLICT_RESOLUTION = new PropertyDescriptor.Builder()
|
||||
.name("conflict-resolution-strategy")
|
||||
.displayName("Conflict Resolution Strategy")
|
||||
.description("Indicates what should happen when a file with the same name already exists in the specified Google Drive folder.")
|
||||
.required(true)
|
||||
.defaultValue(FAIL_RESOLUTION)
|
||||
.allowableValues(FAIL_RESOLUTION, IGNORE_RESOLUTION, REPLACE_RESOLUTION)
|
||||
.defaultValue(FAIL.getValue())
|
||||
.allowableValues(ConflictResolutionStrategy.class)
|
||||
.build();
|
||||
|
||||
public static final PropertyDescriptor CHUNKED_UPLOAD_SIZE = new PropertyDescriptor.Builder()
|
||||
|
|
@ -190,8 +163,6 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
public static final List<PropertyDescriptor> PROPERTIES = Collections.unmodifiableList(asList(
|
||||
GCP_CREDENTIALS_PROVIDER_SERVICE,
|
||||
FOLDER_ID,
|
||||
SUBFOLDER_NAME,
|
||||
CREATE_SUBFOLDER,
|
||||
FILE_NAME,
|
||||
CONFLICT_RESOLUTION,
|
||||
CHUNKED_UPLOAD_THRESHOLD,
|
||||
|
|
@ -260,15 +231,11 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
return;
|
||||
}
|
||||
|
||||
String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
|
||||
final String subfolderName = context.getProperty(SUBFOLDER_NAME).evaluateAttributeExpressions(flowFile).getValue();
|
||||
final boolean createFolder = context.getProperty(CREATE_SUBFOLDER).asBoolean();
|
||||
final String folderId = context.getProperty(FOLDER_ID).evaluateAttributeExpressions(flowFile).getValue();
|
||||
final String filename = context.getProperty(FILE_NAME).evaluateAttributeExpressions(flowFile).getValue();
|
||||
final String mimeType = flowFile.getAttribute(CoreAttributes.MIME_TYPE.key());
|
||||
|
||||
try {
|
||||
folderId = subfolderName != null ? getOrCreateParentSubfolder(subfolderName, folderId, createFolder).getId() : folderId;
|
||||
|
||||
final long startNanos = System.nanoTime();
|
||||
final long size = flowFile.getSize();
|
||||
|
||||
|
|
@ -280,20 +247,19 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
.asDataSize(DataUnit.B)
|
||||
.intValue();
|
||||
|
||||
final String conflictResolution = context.getProperty(CONFLICT_RESOLUTION).getValue();
|
||||
|
||||
final ConflictResolutionStrategy conflictResolution = ConflictResolutionStrategy.forValue(context.getProperty(CONFLICT_RESOLUTION).getValue());
|
||||
final Optional<File> alreadyExistingFile = checkFileExistence(filename, folderId);
|
||||
final File fileMetadata = alreadyExistingFile.isPresent() ? alreadyExistingFile.get() : createMetadata(filename, folderId);
|
||||
final File fileMetadata = alreadyExistingFile.orElseGet(() -> createMetadata(filename, folderId));
|
||||
|
||||
if (alreadyExistingFile.isPresent() && FAIL_RESOLUTION.equals(conflictResolution)) {
|
||||
getLogger().error("File '{}' already exists in {} folder, conflict resolution is '{}'", filename, getFolderName(subfolderName), FAIL_RESOLUTION);
|
||||
if (alreadyExistingFile.isPresent() && conflictResolution == FAIL) {
|
||||
getLogger().error("File [{}] already exists in [{}] Folder, conflict resolution is [{}]", filename, folderId, FAIL.getDisplayName());
|
||||
flowFile = addAttributes(alreadyExistingFile.get(), flowFile, session);
|
||||
session.transfer(flowFile, REL_FAILURE);
|
||||
return;
|
||||
}
|
||||
|
||||
if (alreadyExistingFile.isPresent() && IGNORE_RESOLUTION.equals(conflictResolution)) {
|
||||
getLogger().info("File '{}' already exists in {} folder, conflict resolution is '{}'", filename, getFolderName(subfolderName), IGNORE_RESOLUTION);
|
||||
if (alreadyExistingFile.isPresent() && conflictResolution == IGNORE) {
|
||||
getLogger().info("File [{}] already exists in [{}] Folder, conflict resolution is [{}]", filename, folderId, IGNORE.getDisplayName());
|
||||
flowFile = addAttributes(alreadyExistingFile.get(), flowFile, session);
|
||||
session.transfer(flowFile, REL_SUCCESS);
|
||||
return;
|
||||
|
|
@ -323,12 +289,12 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
}
|
||||
session.transfer(flowFile, REL_SUCCESS);
|
||||
} catch (GoogleJsonResponseException e) {
|
||||
getLogger().error("Exception occurred while uploading file '{}' to {} Google Drive folder", filename,
|
||||
getFolderName(subfolderName), e);
|
||||
getLogger().error("Exception occurred while uploading File [{}] to [{}] Google Drive Folder", filename,
|
||||
folderId, e);
|
||||
handleExpectedError(session, flowFile, e);
|
||||
} catch (Exception e) {
|
||||
getLogger().error("Exception occurred while uploading file '{}' to {} Google Drive folder", filename,
|
||||
getFolderName(subfolderName), e);
|
||||
getLogger().error("Exception occurred while uploading File [{}] to [{}] Google Drive Folder", filename,
|
||||
folderId, e);
|
||||
|
||||
if (e.getCause() != null && e.getCause() instanceof GoogleJsonResponseException) {
|
||||
handleExpectedError(session, flowFile, (GoogleJsonResponseException) e.getCause());
|
||||
|
|
@ -354,10 +320,6 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
return session.putAllAttributes(flowFile, attributes);
|
||||
}
|
||||
|
||||
private String getFolderName(String subFolderName) {
|
||||
return subFolderName == null ? "shared" : format("'%s'", subFolderName);
|
||||
}
|
||||
|
||||
private DriveRequest<File> createDriveRequest(File fileMetadata, final InputStreamContent mediaContent) throws IOException {
|
||||
if (fileMetadata.getId() == null) {
|
||||
return driveService.files()
|
||||
|
|
@ -384,7 +346,8 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
fileMetadata.setSize(mediaContent.getLength());
|
||||
return fileMetadata;
|
||||
} else {
|
||||
throw new ProcessException(format("Upload of file '%s' to folder '%s' failed, HTTP error code: %d", fileMetadata.getName(), fileMetadata.getId(), response.getStatusCode()));
|
||||
throw new ProcessException(format("Upload of File [%s] to Folder [%s] failed, HTTP error code: [%d]",
|
||||
fileMetadata.getName(), fileMetadata.getId(), response.getStatusCode()));
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -395,44 +358,6 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
return new JSONObject(contentAsString).getString("id");
|
||||
}
|
||||
|
||||
private File getOrCreateParentSubfolder(String folderName, String parentFolderId, boolean createFolder) throws IOException {
|
||||
final int indexOfPathSeparator = folderName.indexOf("/");
|
||||
|
||||
if (isMultiLevelFolder(indexOfPathSeparator, folderName)) {
|
||||
final String mainFolderName = folderName.substring(0, indexOfPathSeparator);
|
||||
final String subFolders = folderName.substring(indexOfPathSeparator + 1);
|
||||
final File mainFolder = getOrCreateFolder(mainFolderName, parentFolderId, createFolder);
|
||||
return getOrCreateParentSubfolder(subFolders, mainFolder.getId(), createFolder);
|
||||
} else {
|
||||
return getOrCreateFolder(folderName, parentFolderId, createFolder);
|
||||
}
|
||||
}
|
||||
|
||||
private boolean isMultiLevelFolder(int indexOfPathSeparator, String folderName) {
|
||||
return indexOfPathSeparator > 0 && indexOfPathSeparator < folderName.length() - 1;
|
||||
}
|
||||
|
||||
private File getOrCreateFolder(String folderName, String parentFolderId, boolean createFolder) throws IOException {
|
||||
final Optional<File> existingFolder = checkFolderExistence(folderName, parentFolderId);
|
||||
|
||||
if (existingFolder.isPresent()) {
|
||||
return existingFolder.get();
|
||||
}
|
||||
|
||||
if (createFolder) {
|
||||
getLogger().debug("Create folder " + folderName + " parent id: " + parentFolderId);
|
||||
final File folderMetadata = createMetadata(folderName, parentFolderId);
|
||||
folderMetadata.setMimeType(DRIVE_FOLDER_MIME_TYPE);
|
||||
|
||||
return driveService.files()
|
||||
.create(folderMetadata)
|
||||
.setFields("id, parents")
|
||||
.execute();
|
||||
} else {
|
||||
throw new ProcessException(format("The specified subfolder '%s' does not exist and '%s' is false.", folderName, CREATE_SUBFOLDER.getDisplayName()));
|
||||
}
|
||||
}
|
||||
|
||||
private File createMetadata(final String name, final String parentId) {
|
||||
final File metadata = new File();
|
||||
metadata.setName(name);
|
||||
|
|
@ -440,18 +365,10 @@ public class PutGoogleDrive extends AbstractProcessor implements GoogleDriveTrai
|
|||
return metadata;
|
||||
}
|
||||
|
||||
private Optional<File> checkFolderExistence(String folderName, String parentId) throws IOException {
|
||||
return checkObjectExistence(format("mimeType='%s' and name='%s' and ('%s' in parents)", DRIVE_FOLDER_MIME_TYPE, folderName, parentId));
|
||||
}
|
||||
|
||||
private Optional<File> checkFileExistence(String fileName, String parentId) throws IOException {
|
||||
return checkObjectExistence(format("name='%s' and ('%s' in parents)", fileName, parentId));
|
||||
}
|
||||
|
||||
private Optional<File> checkObjectExistence(String query) throws IOException {
|
||||
final FileList result = driveService.files()
|
||||
.list()
|
||||
.setQ(query)
|
||||
.setQ(format("name='%s' and ('%s' in parents)", fileName, parentId))
|
||||
.setFields("files(name, id)")
|
||||
.execute();
|
||||
|
||||
|
|
|
|||
|
|
@ -42,7 +42,6 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
|
||||
@Test
|
||||
void testFetch() throws Exception {
|
||||
// GIVEN
|
||||
File file = createFileWithDefaultContent("test_file.txt", mainFolderId);
|
||||
|
||||
Map<String, String> inputFlowFileAttributes = new HashMap<>();
|
||||
|
|
@ -54,11 +53,9 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
HashSet<Map<String, String>> expectedAttributes = new HashSet<>(singletonList(inputFlowFileAttributes));
|
||||
List<String> expectedContent = singletonList(DEFAULT_FILE_CONTENT);
|
||||
|
||||
// WHEN
|
||||
testRunner.enqueue("unimportant_data", inputFlowFileAttributes);
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(FetchGoogleDrive.REL_FAILURE, 0);
|
||||
|
||||
checkAttributes(FetchGoogleDrive.REL_SUCCESS, expectedAttributes);
|
||||
|
|
@ -67,7 +64,6 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
|
||||
@Test
|
||||
void testInputFlowFileReferencesMissingFile() {
|
||||
// GIVEN
|
||||
Map<String, String> inputFlowFileAttributes = new HashMap<>();
|
||||
inputFlowFileAttributes.put(GoogleDriveAttributes.ID, "missing");
|
||||
inputFlowFileAttributes.put(GoogleDriveAttributes.FILENAME, "missing_filename");
|
||||
|
|
@ -80,24 +76,20 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
}}
|
||||
));
|
||||
|
||||
// WHEN
|
||||
testRunner.enqueue("unimportant_data", inputFlowFileAttributes);
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(FetchGoogleDrive.REL_SUCCESS, 0);
|
||||
checkAttributes(FetchGoogleDrive.REL_FAILURE, expectedFailureAttributes);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testInputFlowFileThrowsExceptionBeforeFetching() throws Exception {
|
||||
// GIVEN
|
||||
File file = createFileWithDefaultContent("test_file.txt", mainFolderId);
|
||||
|
||||
Map<String, String> inputFlowFileAttributes = new HashMap<>();
|
||||
inputFlowFileAttributes.put(GoogleDriveAttributes.ID, file.getId());
|
||||
inputFlowFileAttributes.put(GoogleDriveAttributes.FILENAME, file.getName());
|
||||
|
||||
MockFlowFile input = new MockFlowFile(1) {
|
||||
final AtomicBoolean throwException = new AtomicBoolean(true);
|
||||
|
||||
|
|
@ -111,7 +103,6 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
return super.isPenalized();
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public Map<String, String> getAttributes() {
|
||||
return inputFlowFileAttributes;
|
||||
|
|
@ -122,11 +113,9 @@ public class FetchGoogleDriveIT extends AbstractGoogleDriveIT<FetchGoogleDrive>
|
|||
inputFlowFileAttributes
|
||||
));
|
||||
|
||||
// WHEN
|
||||
testRunner.enqueue(input);
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(FetchGoogleDrive.REL_SUCCESS, 0);
|
||||
|
||||
checkAttributes(FetchGoogleDrive.REL_FAILURE, expectedFailureAttributes);
|
||||
|
|
|
|||
|
|
@ -49,7 +49,6 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
|
||||
@Test
|
||||
void listFilesFrom3LayerDeepDirectoryTree() throws Exception {
|
||||
// GIVEN
|
||||
File main_sub1 = createFolder("main_sub1", mainFolderId);
|
||||
File main_sub2 = createFolder("main_sub2", mainFolderId);
|
||||
|
||||
|
|
@ -78,10 +77,8 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
// The creation of the files are not (completely) synchronized.
|
||||
Thread.sleep(2000);
|
||||
|
||||
// WHEN
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
List<MockFlowFile> successFlowFiles = testRunner.getFlowFilesForRelationship(ListGoogleDrive.REL_SUCCESS);
|
||||
|
||||
Set<String> actualFileNames = successFlowFiles.stream()
|
||||
|
|
@ -93,19 +90,17 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
// Next, list a sub folder, non-recursively this time. (Changing these properties will clear the Processor state as well
|
||||
// so all files are eligible for listing again.)
|
||||
|
||||
// GIVEN
|
||||
|
||||
testRunner.clearTransferState();
|
||||
|
||||
expectedFileNames = new HashSet<>(Arrays.asList(
|
||||
"main_sub1_file1"
|
||||
));
|
||||
|
||||
// WHEN
|
||||
testRunner.setProperty(ListGoogleDrive.FOLDER_ID, main_sub1.getId());
|
||||
testRunner.setProperty(ListGoogleDrive.RECURSIVE_SEARCH, "false");
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
successFlowFiles = testRunner.getFlowFilesForRelationship(ListGoogleDrive.REL_SUCCESS);
|
||||
|
||||
actualFileNames = successFlowFiles.stream()
|
||||
|
|
@ -117,7 +112,6 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
|
||||
@Test
|
||||
void doNotListTooYoungFilesWhenMinAgeIsSet() throws Exception {
|
||||
// GIVEN
|
||||
testRunner.setProperty(ListGoogleDrive.MIN_AGE, "15 s");
|
||||
|
||||
createFileWithDefaultContent("main_file", mainFolderId);
|
||||
|
|
@ -125,10 +119,8 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
// Make sure the file 'arrives' and could be listed
|
||||
Thread.sleep(5000);
|
||||
|
||||
// WHEN
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
List<MockFlowFile> successFlowFiles = testRunner.getFlowFilesForRelationship(ListGoogleDrive.REL_SUCCESS);
|
||||
|
||||
Set<String> actualFileNames = successFlowFiles.stream()
|
||||
|
|
@ -139,17 +131,15 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
|
||||
// Next, wait for another 10+ seconds for MIN_AGE to expire then list again
|
||||
|
||||
// GIVEN
|
||||
|
||||
Thread.sleep(10000);
|
||||
|
||||
Set<String> expectedFileNames = new HashSet<>(Arrays.asList(
|
||||
"main_file"
|
||||
));
|
||||
|
||||
// WHEN
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
successFlowFiles = testRunner.getFlowFilesForRelationship(ListGoogleDrive.REL_SUCCESS);
|
||||
|
||||
actualFileNames = successFlowFiles.stream()
|
||||
|
|
@ -158,5 +148,4 @@ public class ListGoogleDriveIT extends AbstractGoogleDriveIT<ListGoogleDrive> {
|
|||
|
||||
assertEquals(expectedFileNames, actualFileNames);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,7 +38,6 @@ import static org.mockito.Mockito.when;
|
|||
|
||||
public class ListGoogleDriveSimpleTest {
|
||||
private ListGoogleDrive testSubject;
|
||||
|
||||
private ProcessContext mockProcessContext;
|
||||
private Drive mockDriverService;
|
||||
|
||||
|
|
|
|||
|
|
@ -84,7 +84,6 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
|
||||
@Test
|
||||
void testOutputAsAttributesWhereTimestampIsCreatedTime() throws Exception {
|
||||
// GIVEN
|
||||
String id = "id_1";
|
||||
String filename = "file_name_1";
|
||||
Long size = 125L;
|
||||
|
|
@ -92,14 +91,10 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
Long modifiedTime = null;
|
||||
String mimeType = "mime_type_1";
|
||||
|
||||
// WHEN
|
||||
// THEN
|
||||
testOutputAsAttributes(id, filename, size, createdTime, modifiedTime, mimeType, createdTime);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testOutputAsAttributesWhereTimestampIsModifiedTime() throws Exception {
|
||||
// GIVEN
|
||||
String id = "id_1";
|
||||
String filename = "file_name_1";
|
||||
Long size = 125L;
|
||||
|
|
@ -107,14 +102,11 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
Long modifiedTime = 123456L + 1L;
|
||||
String mimeType = "mime_type_1";
|
||||
|
||||
// WHEN
|
||||
// THEN
|
||||
testOutputAsAttributes(id, filename, size, createdTime, modifiedTime, mimeType, modifiedTime);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testOutputAsContent() throws Exception {
|
||||
// GIVEN
|
||||
String id = "id_1";
|
||||
String filename = "file_name_1";
|
||||
Long size = 125L;
|
||||
|
|
@ -137,10 +129,7 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
"}" +
|
||||
"]");
|
||||
|
||||
// WHEN
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
checkContent(ListGoogleDrive.REL_SUCCESS, expectedContents);
|
||||
}
|
||||
|
||||
|
|
@ -172,7 +161,6 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
}
|
||||
|
||||
private void testOutputAsAttributes(String id, String filename, Long size, Long createdTime, Long modifiedTime, String mimeType, Long expectedTimestamp) throws IOException {
|
||||
// GIVEN
|
||||
mockFetchedGoogleDriveFileList(id, filename, size, createdTime, modifiedTime, mimeType);
|
||||
|
||||
Map<String, String> inputFlowFileAttributes = new HashMap<>();
|
||||
|
|
@ -184,10 +172,8 @@ public class ListGoogleDriveTestRunnerTest implements OutputChecker {
|
|||
|
||||
HashSet<Map<String, String>> expectedAttributes = new HashSet<>(asList(inputFlowFileAttributes));
|
||||
|
||||
// WHEN
|
||||
testRunner.run();
|
||||
|
||||
// THEN
|
||||
checkAttributes(ListGoogleDrive.REL_SUCCESS, expectedAttributes);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -16,12 +16,11 @@
|
|||
*/
|
||||
package org.apache.nifi.processors.gcp.drive;
|
||||
|
||||
import static org.apache.nifi.processors.gcp.drive.PutGoogleDrive.CREATE_SUBFOLDER;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
|
||||
import static org.apache.nifi.processors.gcp.drive.PutGoogleDrive.FILE_NAME;
|
||||
import static org.apache.nifi.processors.gcp.drive.PutGoogleDrive.FOLDER_ID;
|
||||
import static org.apache.nifi.processors.gcp.drive.PutGoogleDrive.SUBFOLDER_NAME;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
|
@ -56,22 +55,6 @@ public class PutGoogleDriveIT extends AbstractGoogleDriveIT<PutGoogleDrive> impl
|
|||
// WHEN
|
||||
runWithFileContent();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_SUCCESS, 1);
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_FAILURE, 0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testUploadFileFolderByName() {
|
||||
// GIVEN
|
||||
testRunner.setProperty(SUBFOLDER_NAME, "testFolderNew");
|
||||
testRunner.setProperty(FOLDER_ID, mainFolderId);
|
||||
testRunner.setProperty(FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(CREATE_SUBFOLDER, "true");
|
||||
|
||||
// WHEN
|
||||
runWithFileContent();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_SUCCESS, 1);
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_FAILURE, 0);
|
||||
|
|
@ -80,47 +63,6 @@ public class PutGoogleDriveIT extends AbstractGoogleDriveIT<PutGoogleDrive> impl
|
|||
assertFlowFileAttributes(ff0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testUploadFileCreateMultiLevelFolder() throws IOException {
|
||||
createFolder("existingFolder", mainFolderId);
|
||||
|
||||
// GIVEN
|
||||
testRunner.setProperty(SUBFOLDER_NAME, "existingFolder/new1/new2");
|
||||
testRunner.setProperty(FOLDER_ID, mainFolderId);
|
||||
testRunner.setProperty(FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(CREATE_SUBFOLDER, "true");
|
||||
|
||||
// WHEN
|
||||
runWithFileContent();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_SUCCESS, 1);
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_FAILURE, 0);
|
||||
|
||||
final List<MockFlowFile> flowFiles = testRunner.getFlowFilesForRelationship(PutGoogleDrive.REL_SUCCESS);
|
||||
final MockFlowFile ff0 = flowFiles.get(0);
|
||||
assertFlowFileAttributes(ff0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testSpecifiedFolderIdDoesNotExist() {
|
||||
// GIVEN
|
||||
testRunner.setProperty(FOLDER_ID, "nonExistentId");
|
||||
testRunner.setProperty(FILE_NAME, "testFile4");
|
||||
|
||||
// WHEN
|
||||
runWithFileContent();
|
||||
|
||||
// THEN
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_SUCCESS, 0);
|
||||
testRunner.assertTransferCount(PutGoogleDrive.REL_FAILURE, 1);
|
||||
|
||||
final List<MockFlowFile> flowFiles = testRunner.getFlowFilesForRelationship(PutGoogleDrive.REL_FAILURE);
|
||||
final MockFlowFile ff0 = flowFiles.get(0);
|
||||
ff0.assertAttributeEquals(GoogleDriveAttributes.ERROR_CODE, "404");
|
||||
ff0.assertAttributeExists(GoogleDriveAttributes.ERROR_MESSAGE);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testUploadedFileAlreadyExistsFailResolution() {
|
||||
// GIVEN
|
||||
|
|
@ -145,11 +87,11 @@ public class PutGoogleDriveIT extends AbstractGoogleDriveIT<PutGoogleDrive> impl
|
|||
}
|
||||
|
||||
@Test
|
||||
void testUploadedFileAlreadyExistsOverwriteResolution() {
|
||||
void testUploadedFileAlreadyExistsReplaceResolution() {
|
||||
// GIVEN
|
||||
testRunner.setProperty(FOLDER_ID, mainFolderId);
|
||||
testRunner.setProperty(FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, PutGoogleDrive.REPLACE_RESOLUTION);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, REPLACE.getValue());
|
||||
|
||||
// WHEN
|
||||
runWithFileContent();
|
||||
|
|
@ -176,7 +118,7 @@ public class PutGoogleDriveIT extends AbstractGoogleDriveIT<PutGoogleDrive> impl
|
|||
// GIVEN
|
||||
testRunner.setProperty(FOLDER_ID, mainFolderId);
|
||||
testRunner.setProperty(FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, PutGoogleDrive.IGNORE_RESOLUTION);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, IGNORE.getValue());
|
||||
|
||||
// WHEN
|
||||
runWithFileContent();
|
||||
|
|
|
|||
|
|
@ -19,9 +19,11 @@ package org.apache.nifi.processors.gcp.drive;
|
|||
|
||||
import static java.lang.String.format;
|
||||
import static java.nio.charset.StandardCharsets.UTF_8;
|
||||
import static java.util.Collections.emptyList;
|
||||
import static java.util.Collections.singletonList;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.IGNORE;
|
||||
import static org.apache.nifi.processors.conflict.resolution.ConflictResolutionStrategy.REPLACE;
|
||||
import static org.apache.nifi.processors.gcp.drive.GoogleDriveAttributes.ERROR_MESSAGE;
|
||||
import static org.apache.nifi.processors.gcp.drive.GoogleDriveTrait.DRIVE_FOLDER_MIME_TYPE;
|
||||
import static org.mockito.ArgumentMatchers.any;
|
||||
import static org.mockito.ArgumentMatchers.eq;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
|
@ -82,26 +84,11 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
testRunner.assertNotValid();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testSubfolderNameValidity() {
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "sub1");
|
||||
testRunner.assertValid();
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "sub1/sub2");
|
||||
testRunner.assertValid();
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "/sub1");
|
||||
testRunner.assertNotValid();
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "/");
|
||||
testRunner.assertNotValid();
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "sub1/");
|
||||
testRunner.assertNotValid();
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, "/sub1/");
|
||||
testRunner.assertNotValid();
|
||||
}
|
||||
|
||||
@Test
|
||||
void testFileUploadFileNameFromProperty() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
|
||||
mockFileExists(emptyList());
|
||||
mockFileUpload(createFile());
|
||||
runWithFlowFile();
|
||||
|
||||
|
|
@ -114,6 +101,7 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
void testFileUploadFileNameFromFlowFileAttribute() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FOLDER_ID, SHARED_FOLDER_ID);
|
||||
|
||||
mockFileExists(emptyList());
|
||||
mockFileUpload(createFile());
|
||||
|
||||
final MockFlowFile mockFlowFile = getMockFlowFile();
|
||||
|
|
@ -128,31 +116,12 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
assertProvenanceEvent(ProvenanceEventType.SEND);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testFileUploadFileToFolderSpecifiedByNameFolderExists() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.SUBFOLDER_NAME, SUBFOLDER_NAME);
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
|
||||
when(mockDriverService.files()
|
||||
.list()
|
||||
.setQ(format("mimeType='%s' and name='%s' and ('%s' in parents)", DRIVE_FOLDER_MIME_TYPE, SUBFOLDER_NAME, SHARED_FOLDER_ID))
|
||||
.setFields("files(name, id)")
|
||||
.execute())
|
||||
.thenReturn(new FileList().setFiles(singletonList(createFile(SUBFOLDER_ID, SUBFOLDER_NAME, SHARED_FOLDER_ID, DRIVE_FOLDER_MIME_TYPE))));
|
||||
|
||||
mockFileUpload(createFile());
|
||||
|
||||
runWithFlowFile();
|
||||
testRunner.assertAllFlowFilesTransferred(PutGoogleDrive.REL_SUCCESS, 1);
|
||||
assertFlowFileAttributes(PutGoogleDrive.REL_SUCCESS);
|
||||
assertProvenanceEvent(ProvenanceEventType.SEND);
|
||||
}
|
||||
|
||||
@Test
|
||||
void testFileUploadError() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
|
||||
final JsonParseException exception = new JsonParseException("Google Drive error", new FileNotFoundException());
|
||||
mockFileExists(emptyList());
|
||||
mockFileUploadError(exception);
|
||||
|
||||
runWithFlowFile();
|
||||
|
|
@ -168,7 +137,7 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
void testFileAlreadyExistsFailResolution() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
|
||||
mockFileExists();
|
||||
mockFileExists(singletonList(createFile()));
|
||||
|
||||
runWithFlowFile();
|
||||
|
||||
|
|
@ -179,9 +148,9 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
@Test
|
||||
void testFileAlreadyExistsIgnoreResolution() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, PutGoogleDrive.IGNORE_RESOLUTION);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, IGNORE.getValue());
|
||||
|
||||
mockFileExists();
|
||||
mockFileExists(singletonList(createFile()));
|
||||
|
||||
runWithFlowFile();
|
||||
|
||||
|
|
@ -193,11 +162,11 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
}
|
||||
|
||||
@Test
|
||||
void testFileAlreadyExistsOverwriteResolution() throws Exception {
|
||||
void testFileAlreadyExistsReplaceResolution() throws Exception {
|
||||
testRunner.setProperty(PutGoogleDrive.FILE_NAME, TEST_FILENAME);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, PutGoogleDrive.REPLACE_RESOLUTION);
|
||||
testRunner.setProperty(PutGoogleDrive.CONFLICT_RESOLUTION, REPLACE.getValue());
|
||||
|
||||
mockFileExists();
|
||||
mockFileExists(singletonList(createFile()));
|
||||
|
||||
mockFileUpdate(createFile());
|
||||
|
||||
|
|
@ -242,12 +211,12 @@ public class PutGoogleDriveTest extends AbstractGoogleDriveTest{
|
|||
.thenThrow(exception);
|
||||
}
|
||||
|
||||
private void mockFileExists() throws IOException {
|
||||
private void mockFileExists(List<File> fileList) throws IOException {
|
||||
when(mockDriverService.files()
|
||||
.list()
|
||||
.setQ(format("name='%s' and ('%s' in parents)", TEST_FILENAME, SHARED_FOLDER_ID))
|
||||
.setFields("files(name, id)")
|
||||
.execute())
|
||||
.thenReturn(new FileList().setFiles(singletonList(createFile())));
|
||||
.thenReturn(new FileList().setFiles(fileList));
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue