NIFI-3442 Explicitly support EL on username, password, port.

Introduced basic test unit

This closes #1474.

Signed-off-by: Andy LoPresto <alopresto@apache.org>

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/pom.xml

@@ -229,6 +229,12 @@ language governing permissions and limitations under the License. -->
             <version>1.4.187</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.mockftpserver</groupId>
+            <artifactId>MockFtpServer</artifactId>
+            <version>2.7.1</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.bazaarvoice.jolt</groupId>
             <artifactId>jolt-core</artifactId>
@@ -280,6 +286,7 @@ language governing permissions and limitations under the License. -->
                 <artifactId>apache-rat-plugin</artifactId>
                 <configuration>
                     <excludes combine.children="append">
+                        <exclude>src/test/resources/randombytes-1</exclude>
                         <exclude>src/test/resources/localhost.cer</exclude>
                         <exclude>src/test/resources/hello.txt</exclude>
                         <exclude>src/test/resources/CharacterSetConversionSamples/Converted.txt</exclude>

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FTPTransfer.java

@@ -74,6 +74,7 @@ public class FTPTransfer implements FileTransfer {
         .addValidator(StandardValidators.PORT_VALIDATOR)
         .required(true)
         .defaultValue("21")
+        .expressionLanguageSupported(true)
         .build();
     public static final PropertyDescriptor PROXY_TYPE = new PropertyDescriptor.Builder()
         .name("Proxy Type")
@@ -547,7 +548,7 @@ public class FTPTransfer implements FileTransfer {
             client.enterLocalPassiveMode();
         }
 
-        final String transferMode = ctx.getProperty(TRANSFER_MODE).evaluateAttributeExpressions(flowFile).getValue();
+        final String transferMode = ctx.getProperty(TRANSFER_MODE).getValue();
         final int fileType = (transferMode.equalsIgnoreCase(TRANSFER_MODE_ASCII)) ? FTPClient.ASCII_FILE_TYPE : FTPClient.BINARY_FILE_TYPE;
         if (!client.setFileType(fileType)) {
             throw new IOException("Unable to set transfer mode to type " + transferMode);

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/main/java/org/apache/nifi/processors/standard/util/FileTransfer.java

@@ -67,6 +67,7 @@ public interface FileTransfer extends Closeable {
         .description("Username")
         .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
         .required(true)
+        .expressionLanguageSupported(true)
         .build();
     public static final PropertyDescriptor PASSWORD = new PropertyDescriptor.Builder()
         .name("Password")
@@ -74,6 +75,7 @@ public interface FileTransfer extends Closeable {
         .addValidator(Validator.VALID)
         .required(false)
         .sensitive(true)
+        .expressionLanguageSupported(true)
         .build();
     public static final PropertyDescriptor DATA_TIMEOUT = new PropertyDescriptor.Builder()
         .name("Data Timeout")

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/java/org/apache/nifi/processors/standard/TestFTP.java

@@ -0,0 +1,170 @@
+/*
+ * 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.standard;
+
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processors.standard.util.FTPTransfer;
+import org.apache.nifi.util.MockFlowFile;
+import org.apache.nifi.util.MockProcessContext;
+import org.apache.nifi.util.TestRunner;
+import org.apache.nifi.util.TestRunners;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockftpserver.fake.FakeFtpServer;
+import org.mockftpserver.fake.UserAccount;
+import org.mockftpserver.fake.filesystem.DirectoryEntry;
+import org.mockftpserver.fake.filesystem.FileEntry;
+import org.mockftpserver.fake.filesystem.FileSystem;
+import org.mockftpserver.fake.filesystem.WindowsFakeFileSystem;
+
+import java.io.FileInputStream;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class TestFTP {
+
+    final FakeFtpServer fakeFtpServer = new FakeFtpServer();
+    final String username = "nifi-ftp-user";
+    final String password = "Test test test chocolate";
+    int ftpPort;
+
+    @Before
+    public void setUp() throws Exception {
+        fakeFtpServer.setServerControlPort(0);
+        fakeFtpServer.addUserAccount(new UserAccount(username, password, "c:\\data"));
+
+        FileSystem fileSystem = new WindowsFakeFileSystem();
+        fileSystem.add(new DirectoryEntry("c:\\data"));
+        fakeFtpServer.setFileSystem(fileSystem);
+
+        fakeFtpServer.start();
+
+        ftpPort = fakeFtpServer.getServerControlPort();
+    }
+
+    @After
+    public void tearDown() throws Exception {
+        fakeFtpServer.stop();
+    }
+
+    @Test
+    public void testValidators() {
+        TestRunner runner = TestRunners.newTestRunner(PutFTP.class);
+        Collection<ValidationResult> results;
+        ProcessContext pc;
+
+        /* Set the basic required values */
+        results = new HashSet<>();
+        runner.setProperty(FTPTransfer.USERNAME, "${el-username}");
+        runner.setProperty(FTPTransfer.HOSTNAME, "static-hostname");
+        runner.setProperty(FTPTransfer.PORT, "${el-portNumber}");
+
+        results = new HashSet<>();
+        runner.setProperty(FTPTransfer.REMOTE_PATH, "static-remote-target");
+        runner.enqueue(new byte[0]);
+        pc = runner.getProcessContext();
+        if (pc instanceof MockProcessContext) {
+            results = ((MockProcessContext) pc).validate();
+        }
+        assertEquals(0, results.size());
+
+
+        results = new HashSet<>();
+        runner.setProperty(FTPTransfer.REMOTE_PATH, "${el-remote-target}");
+        runner.enqueue(new byte[0]);
+        pc = runner.getProcessContext();
+        if (pc instanceof MockProcessContext) {
+            results = ((MockProcessContext) pc).validate();
+        }
+        assertEquals(0, results.size());
+
+        results = new HashSet<>();
+        runner.setProperty(FTPTransfer.USERNAME, "static-username");
+        runner.enqueue(new byte[0]);
+        pc = runner.getProcessContext();
+        if (pc instanceof MockProcessContext) {
+            results = ((MockProcessContext) pc).validate();
+        }
+        assertEquals(0, results.size());
+
+        /* Try an invalid expression */
+        results = new HashSet<>();
+        runner.setProperty(FTPTransfer.USERNAME, "");
+        runner.enqueue(new byte[0]);
+        pc = runner.getProcessContext();
+        if (pc instanceof MockProcessContext) {
+            results = ((MockProcessContext) pc).validate();
+        }
+        assertEquals(1, results.size());
+
+    }
+
+    @Test
+    public void basicFileUpload() throws IOException {
+        TestRunner runner = TestRunners.newTestRunner(PutFTP.class);
+        runner.setProperty(FTPTransfer.HOSTNAME, "localhost");
+        runner.setProperty(FTPTransfer.USERNAME, username);
+        runner.setProperty(FTPTransfer.PASSWORD, password);
+        runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
+        try (FileInputStream fis = new FileInputStream("src/test/resources/randombytes-1");) {
+            Map<String, String> attributes = new HashMap<String, String>();
+            attributes.put(CoreAttributes.FILENAME.key(), "randombytes-1");
+            runner.enqueue(fis, attributes);
+            runner.run();
+        }
+        runner.assertQueueEmpty();
+        runner.assertAllFlowFilesTransferred(PutFTP.REL_SUCCESS);
+        FileSystem results = fakeFtpServer.getFileSystem();
+
+        // Check file was uploaded
+        Assert.assertTrue(results.exists("c:\\data\\randombytes-1"));
+    }
+
+    @Test
+    public void basicFileGet() throws IOException {
+        FileSystem results = fakeFtpServer.getFileSystem();
+
+        FileEntry sampleFile = new FileEntry("c:\\data\\randombytes-2");
+        sampleFile.setContents("Just some random test test test chocolate");
+        results.add(sampleFile);
+
+        // Check file exists
+        Assert.assertTrue(results.exists("c:\\data\\randombytes-2"));
+
+        TestRunner runner = TestRunners.newTestRunner(GetFTP.class);
+        runner.setProperty(FTPTransfer.HOSTNAME, "localhost");
+        runner.setProperty(FTPTransfer.USERNAME, username);
+        runner.setProperty(FTPTransfer.PASSWORD, password);
+        runner.setProperty(FTPTransfer.PORT, Integer.toString(ftpPort));
+        runner.setProperty(FTPTransfer.REMOTE_PATH, "/");
+
+        runner.run();
+
+        final MockFlowFile retrievedFile = runner.getFlowFilesForRelationship(GetFTP.REL_SUCCESS).get(0);
+        retrievedFile.assertContentEquals("Just some random test test test chocolate");
+    }
+}

nifi-nar-bundles/nifi-standard-bundle/nifi-standard-processors/src/test/resources/randombytes-1

@@ -0,0 +1 @@
+test test test chocolate